Seek stream_position error when using Compat with File

[tmpfs]

Version

tokio-compat-seek v0.1.0 (/Users/muji/Desktop/tokio-compat-seek)
├── tokio v1.28.2
│   └── tokio-macros v2.1.0 (proc-macro)
└── tokio-util v0.7.8
    └── tokio v1.28.2 (*)

Platform

Darwin sky 21.6.0 Darwin Kernel Version 21.6.0: Mon Dec 19 20:43:09 PST 2022; root:xnu-8020.240.18~2/RELEASE_ARM64_T6000 arm64

Description

Calling stream_position() after a write fails when using the futures::io compatibility traits but works when using tokio::fs::File directly.

Here is a small program that reproduces the issue:

use anyhow::Result;

async fn tokio_stream_position() -> Result<()> {
    use tokio::fs::File;
    use tokio::io::{AsyncSeekExt, AsyncWriteExt};
    let path = "target/tokio-seek.test";

    let mut file = File::create(path).await?;
    let buffer = [0u8; 16];
    file.write_all(&buffer).await?;
    
    let pos = file.stream_position().await?;
    println!("tokio got pos {}", pos);

    Ok(())
}


async fn compat_stream_position() -> Result<()> {
    use tokio::fs::File;
    use futures::io::{AsyncSeekExt, AsyncWriteExt};
    use tokio_util::compat::TokioAsyncWriteCompatExt;
    let path = "target/compat-seek.test";

    let mut file = File::create(path).await?.compat_write();
    let buffer = [0u8; 16];
    file.write_all(&buffer).await?;
    
    // Error: other file operation is pending, 
    // call poll_complete before start_seek
    let pos = file.stream_position().await?;
    println!("futures got pos {}", pos);

    Ok(())
}

#[tokio::main]
async fn main() -> Result<()> {
    tokio_stream_position().await?;
    compat_stream_position().await?;
    Ok(())
}

Yields this output:

tokio got pos 16
Error: other file operation is pending, call poll_complete before start_seek

Cargo.toml:

[package]
name = "tokio-compat-seek"
version = "0.1.0"
edition = "2021"

[dependencies]
futures = { version = "0.3" }
anyhow = "1"
tokio = { version = "1", default-features = false, features = ["rt", "macros", "fs", "io-util", "rt-multi-thread"] }
tokio-util = { version = "0.7", features = ["compat"] }

I expected to see the correct stream position in both function calls, instead the second call using Compat errors.

Any ideas why this error is happening?

[tmpfs]

MVP is here:

https://github.com/tmpfs/tokio-compat-seek

I notice that if I call flush() after the write_all() in the compat version then the call to stream_position() works as expected.

[Darksonn]

Thank you. This seems like a bug in the Compat utility from the tokio-util crate.

[dhruv9vats]

I'd like to give this a shot, if that's alright?

Any suggestion for a place to start looking at is greatly welcomed.

[Darksonn]

The problem is likely here:

tokio/tokio-util/src/compat.rs

Lines 223 to 237 in a2941e4

	impl<T: tokio::io::AsyncSeek> futures_io::AsyncSeek for Compat<T> {
	fn poll_seek(
	mut self: Pin<&mut Self>,
	cx: &mut Context<'_>,
	pos: io::SeekFrom,
	) -> Poll<io::Result<u64>> {
	if self.seek_pos != Some(pos) {
	self.as_mut().project().inner.start_seek(pos)?;
	*self.as_mut().project().seek_pos = Some(pos);
	}
	let res = ready!(self.as_mut().project().inner.poll_complete(cx));
	*self.as_mut().project().seek_pos = None;
	Poll::Ready(res.map(|p| p as u64))
	}
	}

[matildasmeds]

Looks like this issue is solved in the merged PR #5783, and we can close this issue. 😊