fix(arrow-avro): fix split sync marker assembly and validate per-block sync markers

arrow-avro/src/reader/async_reader/mod.rs

@@ -386,6 +386,11 @@ impl<R: AsyncFileReader + Unpin + 'static> AsyncAvroFileReader<R> {
                     // If we reached the end of the block, flush it, and move to read batches.
                     if let Some(block) = self.block_decoder.flush() {
                         // Successfully decoded a block.
+                        if block.sync != self.sync_marker {
+                            return self.finish_with_error(AvroError::ParseError(
+                                "Avro block sync marker does not match file header".to_string(),
+                            ));
+                        }
                         let block_count = block.count;
                         let block_data = Bytes::from_owner(if let Some(ref codec) = self.codec {
                             match codec.decompress(&block.data) {
@@ -1083,6 +1088,24 @@ mod tests {
         assert!(batch.num_rows() > 0);
     }
 
+    #[tokio::test]
+    async fn test_block_sync_marker_mismatch_errors() {
+        use tempfile::tempdir;
+        let file = arrow_test_data("avro/alltypes_plain.avro");
+        let mut bytes = std::fs::read(&file).unwrap();
+        // The file ends with the final block's 16-byte sync marker.
+        let last = bytes.len() - 1;
+        bytes[last] ^= 0xFF;
+        let dir = tempdir().unwrap();
+        let path = dir.path().join("corrupt_sync.avro");
+        std::fs::write(&path, bytes).unwrap();
+        let schema = get_alltypes_schema();
+        let err = read_async_file(path.to_str().unwrap(), 1024, None, Some(schema), None)
+            .await
+            .expect_err("corrupted block sync marker should fail the read");
+        assert!(err.to_string().contains("sync marker"), "{err}");
+    }
+
     #[tokio::test]
     async fn test_range_no_sync_marker() {
         // Small range unlikely to contain sync marker

arrow-avro/src/reader/block.rs

@@ -118,8 +118,10 @@ impl BlockDecoder {
                 }
                 BlockDecoderState::Sync => {
                     let to_decode = buf.len().min(self.bytes_remaining);
-                    let write = &mut self.in_progress.sync[16 - to_decode..];
-                    write[..to_decode].copy_from_slice(&buf[..to_decode]);
+                    // Fill from the front: the marker may arrive split across decode() calls.
+                    let offset = 16 - self.bytes_remaining;
+                    self.in_progress.sync[offset..offset + to_decode]
+                        .copy_from_slice(&buf[..to_decode]);
                     self.bytes_remaining -= to_decode;
                     buf = &buf[to_decode..];
                     if self.bytes_remaining == 0 {
@@ -222,4 +224,23 @@ mod tests {
         assert_eq!(block.data, payload);
         assert_eq!(block.sync, sync);
     }
+
+    #[test]
+    fn test_sync_marker_split_across_decode_calls() {
+        // count=1 (zig-zag 0x02), size=1 (0x02), one data byte, 16-byte sync marker
+        let sync: [u8; 16] = core::array::from_fn(|i| i as u8);
+        let mut block_bytes = vec![0x02, 0x02, 0xAA];
+        block_bytes.extend_from_slice(&sync);
+
+        for chunk_size in 1..block_bytes.len() {
+            let mut decoder = BlockDecoder::default();
+            for chunk in block_bytes.chunks(chunk_size) {
+                decoder.decode(chunk).unwrap();
+            }
+            let block = decoder.flush().expect("complete block");
+            assert_eq!(block.count, 1);
+            assert_eq!(block.data, vec![0xAA]);
+            assert_eq!(block.sync, sync, "chunk_size {chunk_size}");
+        }
+    }
 }

arrow-avro/src/reader/header.rs

@@ -314,8 +314,9 @@ impl HeaderDecoder {
                 }
                 HeaderDecoderState::Sync => {
                     let to_decode = buf.len().min(self.bytes_remaining);
-                    let write = &mut self.sync_marker[16 - to_decode..];
-                    write[..to_decode].copy_from_slice(&buf[..to_decode]);
+                    // Fill from the front: the marker may arrive split across decode() calls.
+                    let offset = 16 - self.bytes_remaining;
+                    self.sync_marker[offset..offset + to_decode].copy_from_slice(&buf[..to_decode]);
                     self.bytes_remaining -= to_decode;
                     buf = &buf[to_decode..];
                     if self.bytes_remaining == 0 {

arrow-avro/src/reader/mod.rs

@@ -1370,6 +1370,11 @@ impl<R: BufRead> Reader<R> {
                 self.reader.consume(consumed);
                 if let Some(block) = self.block_decoder.flush() {
                     // Successfully decoded a block.
+                    if block.sync != self.header.sync() {
+                        return Err(AvroError::ParseError(
+                            "Avro block sync marker does not match file header".to_string(),
+                        ));
+                    }
                     self.block_data = if let Some(ref codec) = self.header.compression()? {
                         let decompressed: Vec<u8> = codec.decompress(&block.data)?;
                         decompressed
@@ -1491,6 +1496,23 @@ mod test {
         arrow::compute::concat_batches(&schema, &batches).unwrap()
     }
 
+    #[test]
+    fn test_block_sync_marker_mismatch_errors() {
+        let path = arrow_test_data("avro/alltypes_plain.avro");
+        let mut bytes = std::fs::read(&path).unwrap();
+        // The file ends with the final block's 16-byte sync marker.
+        let last = bytes.len() - 1;
+        bytes[last] ^= 0xFF;
+        let reader = ReaderBuilder::new()
+            .with_batch_size(1024)
+            .build(std::io::Cursor::new(bytes))
+            .unwrap();
+        let err = reader
+            .collect::<Result<Vec<_>, _>>()
+            .expect_err("corrupted block sync marker should fail the read");
+        assert!(err.to_string().contains("sync marker"), "{err}");
+    }
+
     fn read_file_strict(
         path: &str,
         batch_size: usize,