feat(oneshot channel): ensure msg won't be dropped on sender side when send returns ok

tokio/Cargo.toml

@@ -92,6 +92,7 @@ time = []
 tokio-macros = { version = "~2.2.0", path = "../tokio-macros", optional = true }
 
 pin-project-lite = "0.2.11"
+rustversion = { git = "https://github.com/wenym1/rustversion", rev = "0b11410" }
 
 # Everything else is optional...
 bytes = { version = "1.0.0", optional = true }

tokio/src/sync/oneshot.rs

@@ -627,7 +627,42 @@ impl<T> Sender<T> {
             )
         });
 
-        Ok(())
+        #[rustversion::since(1.70)]
+        fn consume_inner<T>(inner: Arc<Inner<T>>) -> Result<(), T> {
+            #[cfg(not(loom))]
+            {
+                if let Some(inner) = Arc::into_inner(inner) {
+                    if let Some(t) = inner.value.with_mut(|ptr| unsafe {
+                        // SAFETY: we have successfully returned with `Some`, which means we are the
+                        // only accessor to `ptr`.
+                        //
+                        // Note: value can be `None` even though we have previously set it as `Some`,
+                        // because the value may have been consumed by receiver before we reach here.
+                        (*ptr).take()
+                    }) {
+                        Err(t)
+                    } else {
+                        Ok(())
+                    }
+                } else {
+                    Ok(())
+                }
+            }
+
+            #[cfg(loom)]
+            {
+                drop(inner);
+                // The `loom::sync::Arc` does not implement `into_inner` yet.
+                Ok(())
+            }
+        }
+
+        #[rustversion::before(1.70)]
+        fn consume_inner<T>(_inner: Arc<Inner<T>>) -> Result<(), T> {
+            Ok(())
+        }
+
+        consume_inner(inner)
     }
 
     /// Waits for the associated [`Receiver`] handle to close.