io: fix futures_io::AsyncSeek implementaion for Compat (#5783)

tokio-util/Cargo.toml

@@ -56,6 +56,7 @@ async-stream = "0.3.0"
 futures = "0.3.0"
 futures-test = "0.3.5"
 parking_lot = "0.12.0"
+tempfile = "3.1.0"
 
 [package.metadata.docs.rs]
 all-features = true

tokio-util/src/compat.rs

@@ -227,6 +227,8 @@ impl<T: tokio::io::AsyncSeek> futures_io::AsyncSeek for Compat<T> {
         pos: io::SeekFrom,
     ) -> Poll<io::Result<u64>> {
         if self.seek_pos != Some(pos) {
+            // Ensure previous seeks have finished before starting a new one
+            ready!(self.as_mut().project().inner.poll_complete(cx))?;
             self.as_mut().project().inner.start_seek(pos)?;
             *self.as_mut().project().seek_pos = Some(pos);
         }

tokio-util/tests/compat.rs

@@ -0,0 +1,43 @@
+#![cfg(all(feature = "compat"))]
+#![cfg(not(target_os = "wasi"))] // WASI does not support all fs operations
+#![warn(rust_2018_idioms)]
+
+use futures_io::SeekFrom;
+use futures_util::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt};
+use tempfile::NamedTempFile;
+use tokio::fs::OpenOptions;
+use tokio_util::compat::TokioAsyncWriteCompatExt;
+
+#[tokio::test]
+async fn compat_file_seek() -> futures_util::io::Result<()> {
+    let temp_file = NamedTempFile::new()?;
+    let mut file = OpenOptions::new()
+        .read(true)
+        .write(true)
+        .create(true)
+        .open(temp_file)
+        .await?
+        .compat_write();
+
+    file.write_all(&[0, 1, 2, 3, 4, 5]).await?;
+    file.write_all(&[6, 7]).await?;
+
+    assert_eq!(file.stream_position().await?, 8);
+
+    // Modify elements at position 2.
+    assert_eq!(file.seek(SeekFrom::Start(2)).await?, 2);
+    file.write_all(&[8, 9]).await?;
+
+    file.flush().await?;
+
+    // Verify we still have 8 elements.
+    assert_eq!(file.seek(SeekFrom::End(0)).await?, 8);
+    // Seek back to the start of the file to read and verify contents.
+    file.seek(SeekFrom::Start(0)).await?;
+
+    let mut buf = Vec::new();
+    let num_bytes = file.read_to_end(&mut buf).await?;
+    assert_eq!(&buf[..num_bytes], &[0, 1, 8, 9, 4, 5, 6, 7]);
+
+    Ok(())
+}