From 8f010120710c376585023732742c5840ae2fd6f1 Mon Sep 17 00:00:00 2001
From: Zhengchao An <anzhengchao@gmail.com>
Date: Thu, 2 Jul 2026 04:00:28 +0800
Subject: [PATCH] fix(replication): address resync review follow-up (#4155)

* fix(replication): address resync crate review

* fix(replication): skip fixext fields correctly

* fix(replication): satisfy resync clippy
---
 crates/replication/src/resync.rs              | 51 +++++++++++++++----
 scripts/check_architecture_migration_rules.sh |  2 +-
 2 files changed, 43 insertions(+), 10 deletions(-)

diff --git a/crates/replication/src/resync.rs b/crates/replication/src/resync.rs
index a75a79ad8..3d2b350e2 100644
--- a/crates/replication/src/resync.rs
+++ b/crates/replication/src/resync.rs
@@ -447,19 +447,28 @@ fn skip_msgp_value<R: Read>(rd: &mut R) -> Result<()> {
             }
             return Ok(());
         }
-        Marker::FixExt1 => 1,
-        Marker::FixExt2 => 2,
-        Marker::FixExt4 => 4,
-        Marker::FixExt8 => 8,
-        Marker::FixExt16 => 16,
+        Marker::FixExt1 => 1 + 1,
+        Marker::FixExt2 => 1 + 2,
+        Marker::FixExt4 => 1 + 4,
+        Marker::FixExt8 => 1 + 8,
+        Marker::FixExt16 => 1 + 16,
         Marker::Ext8 => 1 + read_skip_len(rd, 1)?,
-        Marker::Ext16 => 2 + read_skip_len(rd, 2)?,
-        Marker::Ext32 => 4 + read_skip_len(rd, 4)?,
+        Marker::Ext16 => 1 + read_skip_len(rd, 2)?,
+        Marker::Ext32 => 1 + read_skip_len(rd, 4)?,
         Marker::Reserved => 0,
     };
     if skip_len > 0 {
-        let mut buf = vec![0u8; skip_len];
-        rd.read_exact(&mut buf)?;
+        skip_exact(rd, skip_len)?;
+    }
+    Ok(())
+}
+
+fn skip_exact<R: Read>(rd: &mut R, mut len: usize) -> Result<()> {
+    let mut buf = [0u8; 1024];
+    while len > 0 {
+        let take = len.min(buf.len());
+        rd.read_exact(&mut buf[..take])?;
+        len -= take;
     }
     Ok(())
 }
@@ -533,4 +542,28 @@ mod tests {
         assert_eq!(got.targets_map["arn:replication:a"].resync_status, ResyncStatusType::ResyncStarted);
         assert_eq!(got.targets_map["arn:replication:a"].replicated_count, 7);
     }
+
+    #[test]
+    fn skip_msgp_value_consumes_ext_type_and_payload() {
+        let mut ext16 = Cursor::new(vec![0xc8, 0, 2, MSGP_TIME_EXT_TYPE as u8, 0xaa, 0xbb, 0x01]);
+        skip_msgp_value(&mut ext16).expect("ext16 should skip");
+        assert!(matches!(rmp::decode::read_marker(&mut ext16), Ok(Marker::FixPos(1))));
+
+        let mut ext32 = Cursor::new(vec![0xc9, 0, 0, 0, 2, MSGP_TIME_EXT_TYPE as u8, 0xaa, 0xbb, 0x01]);
+        skip_msgp_value(&mut ext32).expect("ext32 should skip");
+        assert!(matches!(rmp::decode::read_marker(&mut ext32), Ok(Marker::FixPos(1))));
+    }
+
+    #[test]
+    fn skip_msgp_value_consumes_fixext_type_and_payload() {
+        for (marker, payload_len) in [(0xd4, 1), (0xd5, 2), (0xd6, 4), (0xd7, 8), (0xd8, 16)] {
+            let mut data = vec![marker, MSGP_TIME_EXT_TYPE as u8];
+            data.resize(data.len() + payload_len, 0xaa);
+            data.push(0x01);
+
+            let mut cursor = Cursor::new(data);
+            skip_msgp_value(&mut cursor).expect("fixext should skip");
+            assert!(matches!(rmp::decode::read_marker(&mut cursor), Ok(Marker::FixPos(1))));
+        }
+    }
 }
diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh
index 850a968fd..a9ff37bc0 100755
--- a/scripts/check_architecture_migration_rules.sh
+++ b/scripts/check_architecture_migration_rules.sh
@@ -2543,7 +2543,7 @@ fi
 
 (
   cd "$ROOT_DIR"
-  rg -n --with-filename 'pub\s+(?:struct|enum)\s+(ResyncOpts|TargetReplicationResyncStatus|BucketReplicationResyncStatus|ResyncStatusType)\b' \
+  rg -n --with-filename 'pub\s+(struct|enum)\s+(ResyncOpts|TargetReplicationResyncStatus|BucketReplicationResyncStatus|ResyncStatusType)\b' \
     crates/ecstore/src/bucket/replication \
     --glob '*.rs' || true
 ) >"$REPLICATION_RESYNC_CONTRACT_BACKSLIDE_HITS_FILE"