diff --git a/Cargo.lock b/Cargo.lock
index 27cf5957fd4..7b2ff8c247c 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -6109,12 +6109,6 @@ version = "1.70.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe"
 
-[[package]]
-name = "oneshot"
-version = "0.2.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cfe21416a02c693fb9f980befcb230ecc70b0b3d1cc4abf88b9675c4c1457f0c"
-
 [[package]]
 name = "onpair"
 version = "0.2.0"
@@ -10116,7 +10110,6 @@ dependencies = [
  "kanal",
  "moka",
  "object_store",
- "oneshot",
  "parking_lot",
  "pin-project-lite",
  "rstest",
@@ -10247,7 +10240,6 @@ dependencies = [
  "itertools 0.14.0",
  "kanal",
  "object_store",
- "oneshot",
  "parking_lot",
  "pin-project-lite",
  "rstest",
@@ -10337,7 +10329,6 @@ dependencies = [
  "kanal",
  "moka",
  "once_cell",
- "oneshot",
  "parking_lot",
  "paste",
  "pin-project-lite",
diff --git a/Cargo.toml b/Cargo.toml
index 00a45b03618..05cf27dccf6 100644
--- a/Cargo.toml
+++ b/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"
diff --git a/vortex-file/Cargo.toml b/vortex-file/Cargo.toml
index 5992ff06434..91c9f5ca8b3 100644
--- a/vortex-file/Cargo.toml
+++ b/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 }
diff --git a/vortex-file/src/read/driver.rs b/vortex-file/src/read/driver.rs
index 5c21da0ba57..a872780f273 100644
--- a/vortex-file/src/read/driver.rs
+++ b/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;
diff --git a/vortex-file/src/read/request.rs b/vortex-file/src/read/request.rs
index c4bb4bdc975..1ef5beb3757 100644
--- a/vortex-file/src/read/request.rs
+++ b/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);
         }
     }
 }
diff --git a/vortex-file/src/segments/source.rs b/vortex-file/src/segments/source.rs
index 3af33362b05..2a985896e08 100644
--- a/vortex-file/src/segments/source.rs
+++ b/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>,
diff --git a/vortex-io/Cargo.toml b/vortex-io/Cargo.toml
index b3f6448484a..cf931d01888 100644
--- a/vortex-io/Cargo.toml
+++ b/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 = [
diff --git a/vortex-io/src/runtime/handle.rs b/vortex-io/src/runtime/handle.rs
index bebbaac5f97..7f768416ec9 100644
--- a/vortex-io/src/runtime/handle.rs
+++ b/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,7 +155,7 @@ 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.
@@ -162,7 +163,7 @@ impl Handle {
             }
         }));
         Task {
-            recv: recv.into_future(),
+            recv,
             abort_handle: Some(abort_handle),
         }
     }
@@ -178,7 +179,7 @@ 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.
@@ -186,7 +187,7 @@ impl Handle {
             }
         }));
         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,
         };
 
diff --git a/vortex-io/src/runtime/single.rs b/vortex-io/src/runtime/single.rs
index 2a113e3715e..21a017ce842 100644
--- a/vortex-io/src/runtime/single.rs
+++ b/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()
         }
     }
diff --git a/vortex-io/src/runtime/tests.rs b/vortex-io/src/runtime/tests.rs
index 940566b4fc1..4730d0350c8 100644
--- a/vortex-io/src/runtime/tests.rs
+++ b/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;
diff --git a/vortex-layout/Cargo.toml b/vortex-layout/Cargo.toml
index f772b9ab639..15729be3ebf 100644
--- a/vortex-layout/Cargo.toml
+++ b/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 }
diff --git a/vortex-layout/src/layouts/dict/writer.rs b/vortex-layout/src/layouts/dict/writer.rs
index b7c93a992fb..878a83f54c9 100644
--- a/vortex-layout/src/layouts/dict/writer.rs
+++ b/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;