replace faern::oneshot with tokio oneshot

Cargo.toml

@@ -196,7 +196,6 @@ num_enum = { version = "0.7.3", default-features = false }
 object_store = { version = "0.13.2", default-features = false }
 object_store_opendal = "0.58.0"
 once_cell = "1.21"
-oneshot = { version = "0.2.0", features = ["async"] }
 onpair = "0.2.0"
 opendal = { version = "0.58.1", default-features = false }
 opentelemetry = "0.32.0"

vortex-file/Cargo.toml

@@ -25,7 +25,6 @@ itertools = { workspace = true }
 kanal = { workspace = true }
 moka = { workspace = true, features = ["sync"] }
 object_store = { workspace = true, optional = true }
-oneshot.workspace = true
 parking_lot = { workspace = true }
 pin-project-lite = { workspace = true }
 tokio = { workspace = true, features = ["rt"], optional = true }

vortex-file/src/read/driver.rs

@@ -137,7 +137,7 @@ impl State {
         trace!(?event, "Received ReadEvent");
         match event {
             ReadEvent::Request(req) => {
-                if req.callback.is_closed() {
+                if req.callback.is_canceled() {
                     trace!(?req, "ReadRequest dropped before registration");
                     return;
                 }
@@ -146,7 +146,7 @@ impl State {
             }
             ReadEvent::Polled(req_id) => {
                 if let Some(req) = self.requests.remove(&req_id) {
-                    if req.callback.is_closed() {
+                    if req.callback.is_canceled() {
                         self.requests_by_offset.remove(&(req.offset, req_id));
                         trace!(?req, "ReadRequest dropped before poll");
                     } else {
@@ -193,7 +193,7 @@ impl State {
     fn next_uncoalesced(&mut self) -> Option<ReadRequest> {
         while let Some((req_id, req)) = self.polled_requests.pop_first() {
             self.requests_by_offset.remove(&(req.offset, req_id));
-            if req.callback.is_closed() {
+            if req.callback.is_canceled() {
                 trace!("Dropping canceled request");
                 continue;
             }
@@ -251,7 +251,7 @@ impl State {
                     .vortex_expect("Missing request in requests_by_offset");
 
                 // Skip any cancelled requests
-                if req.callback.is_closed() {
+                if req.callback.is_canceled() {
                     if ids_to_remove.insert(req_id) {
                         keys_to_remove.push((req_offset, req_id));
                     }
@@ -326,6 +326,7 @@ impl State {
 #[cfg(test)]
 mod tests {
     use futures::StreamExt;
+    use futures::channel::oneshot;
     use futures::stream;
     use vortex_array::buffer::BufferHandle;
     use vortex_buffer::Alignment;

vortex-file/src/read/request.rs

@@ -7,6 +7,7 @@ use std::fmt::Formatter;
 use std::ops::Range;
 use std::sync::Arc;
 
+use futures::channel::oneshot;
 use tracing::trace;
 use vortex_array::buffer::BufferHandle;
 use vortex_buffer::Alignment;
@@ -106,15 +107,15 @@ impl Debug for ReadRequest {
             .field("offset", &self.offset)
             .field("length", &self.length)
             .field("alignment", &self.alignment)
-            .field("is_closed", &self.callback.is_closed())
+            .field("is_canceled", &self.callback.is_canceled())
             .finish()
     }
 }
 
 impl ReadRequest {
     pub(crate) fn resolve(self, result: VortexResult<BufferHandle>) {
-        if let Err(e) = self.callback.send(result) {
-            trace!("ReadRequest {} dropped before resolving: {e}", self.id);
+        if self.callback.send(result).is_err() {
+            trace!("ReadRequest {} dropped before resolving", self.id);
         }
     }
 }

vortex-file/src/segments/source.rs

@@ -13,6 +13,7 @@ use std::task::Poll;
 use futures::FutureExt;
 use futures::StreamExt;
 use futures::channel::mpsc;
+use futures::channel::oneshot;
 use futures::future;
 use futures::future::BoxFuture;
 use futures::future::Shared;
@@ -229,7 +230,7 @@ impl SegmentSource for FileSegmentSource {
 
         let fut = ReadFuture {
             id,
-            recv: recv.into_future(),
+            recv,
             polled: false,
             finished: false,
             events: self.events.clone(),
@@ -248,7 +249,7 @@ impl SegmentSource for FileSegmentSource {
 /// If dropped, the read request will be canceled where possible.
 struct ReadFuture {
     id: usize,
-    recv: oneshot::AsyncReceiver<VortexResult<BufferHandle>>,
+    recv: oneshot::Receiver<VortexResult<BufferHandle>>,
     polled: bool,
     finished: bool,
     events: mpsc::UnboundedSender<ReadEvent>,

vortex-io/Cargo.toml

@@ -25,7 +25,6 @@ futures = { workspace = true, features = ["std", "executor"] }
 glob = { workspace = true }
 kanal = { workspace = true }
 object_store = { workspace = true, optional = true, features = ["fs"] }
-oneshot = { workspace = true }
 parking_lot = { workspace = true }
 pin-project-lite = { workspace = true }
 tokio = { workspace = true, features = [

vortex-io/src/runtime/handle.rs

@@ -11,6 +11,7 @@ use std::task::Poll;
 use std::task::ready;
 
 use futures::FutureExt;
+use futures::channel::oneshot;
 use tracing::Instrument;
 use vortex_error::vortex_panic;
 
@@ -89,7 +90,7 @@ impl Handle {
             .boxed(),
         );
         Task {
-            recv: recv.into_future(),
+            recv,
             abort_handle: Some(abort_handle),
         }
     }
@@ -130,7 +131,7 @@ impl Handle {
             .boxed(),
         );
         Task {
-            recv: recv.into_future(),
+            recv,
             abort_handle: Some(abort_handle),
         }
     }
@@ -154,15 +155,15 @@ impl Handle {
         let abort_handle = self.runtime().spawn_cpu(Box::new(move || {
             let _guard = span.enter();
             // Optimistically avoid the work if the result won't be used.
-            if !send.is_closed() {
+            if !send.is_canceled() {
                 // Catch a panic so it re-raises on the joining side (see `Task::poll`).
                 let output = std::panic::catch_unwind(AssertUnwindSafe(f));
                 // Task::detach allows the receiver to be dropped, so we ignore send errors.
                 drop(send.send(output));
             }
         }));
         Task {
-            recv: recv.into_future(),
+            recv,
             abort_handle: Some(abort_handle),
         }
     }
@@ -178,15 +179,15 @@ impl Handle {
         let abort_handle = self.runtime().spawn_blocking_io(Box::new(move || {
             let _guard = span.enter();
             // Optimistically avoid the work if the result won't be used.
-            if !send.is_closed() {
+            if !send.is_canceled() {
                 // Catch a panic so it re-raises on the joining side (see `Task::poll`).
                 let output = std::panic::catch_unwind(AssertUnwindSafe(f));
                 // Task::detach allows the receiver to be dropped, so we ignore send errors.
                 drop(send.send(output));
             }
         }));
         Task {
-            recv: recv.into_future(),
+            recv,
             abort_handle: Some(abort_handle),
         }
     }
@@ -215,7 +216,7 @@ pub enum JoinOutcome<T> {
 /// continue running in the background, call [`Task::detach`].
 #[must_use = "When a Task is dropped without being awaited, it is cancelled"]
 pub struct Task<T> {
-    recv: oneshot::AsyncReceiver<TaskOutput<T>>,
+    recv: oneshot::Receiver<TaskOutput<T>>,
     abort_handle: Option<AbortHandleRef>,
 }
 
@@ -292,7 +293,7 @@ mod tests {
         drop(send);
 
         let mut task = Task::<()> {
-            recv: recv.into_future(),
+            recv,
             abort_handle: None,
         };
 
@@ -312,7 +313,7 @@ mod tests {
         drop(send.send(Ok(7)));
 
         let mut task = Task::<u32> {
-            recv: recv.into_future(),
+            recv,
             abort_handle: None,
         };
 

vortex-io/src/runtime/single.rs

@@ -7,6 +7,7 @@ use std::sync::Arc;
 
 use futures::Stream;
 use futures::StreamExt;
+use futures::channel::oneshot;
 use futures::future::BoxFuture;
 use futures::stream::LocalBoxStream;
 use parking_lot::Mutex;
@@ -236,7 +237,7 @@ struct LazyAbortHandle {
 impl AbortHandle for LazyAbortHandle {
     fn abort(self: Box<Self>) {
         // Aborting a smol::Task is done by dropping it.
-        if let Ok(task) = self.task.lock().try_recv() {
+        if let Ok(Some(task)) = self.task.lock().try_recv() {
             task.abort()
         }
     }

vortex-io/src/runtime/tests.rs

@@ -10,6 +10,7 @@ use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::Ordering;
 
 use futures::FutureExt;
+use futures::channel::oneshot;
 use futures::future::BoxFuture;
 use tempfile::NamedTempFile;
 use vortex_array::buffer::BufferHandle;

vortex-layout/Cargo.toml

@@ -29,7 +29,6 @@ itertools = { workspace = true }
 kanal = { workspace = true }
 moka = { workspace = true, features = ["future"] }
 once_cell = { workspace = true, features = ["parking_lot"] }
-oneshot = { workspace = true }
 parking_lot = { workspace = true }
 paste = { workspace = true }
 pin-project-lite = { workspace = true }

vortex-layout/src/layouts/dict/writer.rs

@@ -13,6 +13,7 @@ use futures::FutureExt;
 use futures::Stream;
 use futures::StreamExt;
 use futures::TryStreamExt;
+use futures::channel::oneshot;
 use futures::future::BoxFuture;
 use futures::pin_mut;
 use futures::stream::BoxStream;