fix(core): clamp chunk size of 0 to 1 for reader and writer

core/core/src/raw/ops.rs

@@ -659,7 +659,7 @@ impl From<options::ReadOptions> for (BytesRange, OpRead, OpReader) {
         let range = value.range;
         let reader = OpReader {
             concurrent: value.concurrent.max(1),
-            chunk: value.chunk,
+            chunk: value.chunk.map(|v| v.max(1)),
             gap: value.gap,
             prefetch: 0,
         };
@@ -671,7 +671,7 @@ impl From<options::ReaderOptions> for (OpRead, OpReader) {
     fn from(value: options::ReaderOptions) -> Self {
         let reader = OpReader {
             concurrent: value.concurrent.max(1),
-            chunk: value.chunk,
+            chunk: value.chunk.map(|v| v.max(1)),
             gap: value.gap,
             prefetch: value.prefetch,
         };
@@ -888,7 +888,7 @@ impl OpWrite {
         }
 
         let writer = OpWriter {
-            chunk: options.chunk,
+            chunk: options.chunk.map(|v| v.max(1)),
         };
         let user_metadata = options.user_metadata.take().map(sorted_user_metadata);
         let mut flags = 0;
@@ -1057,7 +1057,7 @@ impl OpWriter {
     /// operations like multipart uploads. So the chunk size may be larger than
     /// the given buffer size.
     pub fn with_chunk(mut self, chunk: usize) -> Self {
-        self.chunk = Some(chunk);
+        self.chunk = Some(chunk.max(1));
         self
     }
 }

core/core/src/types/read/reader.rs

@@ -775,6 +775,23 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn test_reader_read_with_chunk_zero() -> Result<()> {
+        let op = Operator::new(services::Memory::default()).unwrap();
+        let path = "test_file";
+
+        let content = gen_fixed_bytes(1024);
+        op.write(path, content.clone())
+            .await
+            .expect("write must succeed");
+
+        let reader = op.reader_with(path).chunk(0).await.unwrap();
+        let buf = reader.read(..).await.expect("read to end must succeed");
+
+        assert_eq!(buf.to_bytes(), content);
+        Ok(())
+    }
+
     #[tokio::test]
     async fn test_reader_read_with_concurrent() -> Result<()> {
         let op = Operator::via_iter(services::MEMORY_SCHEME, [])?;
@@ -966,6 +983,27 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn test_fetch_with_chunk_zero() -> Result<()> {
+        let op = Operator::new(services::Memory::default()).unwrap();
+        let path = "test_file";
+
+        let content = gen_fixed_bytes(1024);
+        op.write(path, content.clone())
+            .await
+            .expect("write must succeed");
+
+        let reader = op.reader_with(path).chunk(0).await.unwrap();
+        let bufs = reader
+            .fetch(vec![0..100, 200..300])
+            .await
+            .expect("fetch must succeed");
+
+        assert_eq!(bufs[0].to_bytes(), content[0..100]);
+        assert_eq!(bufs[1].to_bytes(), content[200..300]);
+        Ok(())
+    }
+
     #[tokio::test]
     async fn test_fetch_empty_ranges() -> Result<()> {
         let op = Operator::new(services::Memory::default()).unwrap();

core/core/src/types/write/writer.rs

@@ -455,6 +455,22 @@ mod tests {
         assert_eq!(buf.to_bytes(), content);
     }
 
+    #[tokio::test]
+    async fn test_writer_write_with_chunk_zero() {
+        let op = Operator::new(services::Memory::default()).unwrap();
+        let path = "test_file";
+
+        let content = vec![b'x'; 4096];
+        op.write_with(path, content.clone())
+            .chunk(0)
+            .await
+            .expect("write with chunk 0 must succeed");
+
+        let buf = op.read(path).await.expect("read to end mut succeed");
+
+        assert_eq!(buf.to_bytes(), content);
+    }
+
     #[tokio::test]
     async fn test_writer_write_from() {
         let op = Operator::new(services::Memory::default()).unwrap();