Prepare Tokio v1.43.3

.github/workflows/ci.yml

@@ -991,10 +991,10 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
-      - name: Install Rust 1.81
+      - name: Install Rust 1.88.0
         uses: dtolnay/rust-toolchain@stable
         with:
-          toolchain: 1.81
+          toolchain: 1.88.0
       - name: Install wasm-pack
         uses: taiki-e/install-action@wasm-pack
 
@@ -1014,10 +1014,10 @@ jobs:
           - wasm32-wasip1-threads
     steps:
       - uses: actions/checkout@v4
-      - name: Install Rust ${{ env.rust_stable }}
+      - name: Install Rust 1.88.0
         uses: dtolnay/rust-toolchain@stable
         with:
-          toolchain: ${{ env.rust_stable }}
+          toolchain: 1.88.0
           targets: ${{ matrix.target }}
 
       # Install dependencies

README.md

@@ -56,7 +56,7 @@ Make sure you activated the full features of the tokio crate on Cargo.toml:
 
 ```toml
 [dependencies]
-tokio = { version = "1.43.2", features = ["full"] }
+tokio = { version = "1.43.3", features = ["full"] }
 ```
 Then, on your main.rs:
 

tests-build/tests/fail/macros_type_mismatch.stderr

@@ -25,7 +25,7 @@ error[E0308]: mismatched types
                    found enum `Result<(), _>`
 help: a return type might be missing here
    |
-9  | async fn missing_return_type() -> _ {
+ 9 | async fn missing_return_type() -> _ {
    |                                ++++
 help: consider using `Result::expect` to unwrap the `Result<(), _>` value, panicking if the value is a `Result::Err`
    |

tokio/CHANGELOG.md

@@ -1,3 +1,15 @@
+# 1.43.3 (October 14th, 2025)
+
+### Fixed
+
+- runtime: use release ordering in `wake_by_ref()` even if already woken ([#7622])
+- sync: close the `broadcast::Sender` in `broadcast::Sender::new()` ([#7629])
+- process: fix error when runtime is shut down on nightly-2025-10-12 ([#7672])
+
+[#7622]: https://github.com/tokio-rs/tokio/pull/7622
+[#7629]: https://github.com/tokio-rs/tokio/pull/7629
+[#7672]: https://github.com/tokio-rs/tokio/pull/7672
+
 # 1.43.2 (August 1st, 2025)
 
 ### Fixed

tokio/Cargo.toml

@@ -6,7 +6,7 @@ name = "tokio"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v1.x.y" git tag.
-version = "1.43.2"
+version = "1.43.3"
 edition = "2021"
 rust-version = "1.70"
 authors = ["Tokio Contributors <team@tokio.rs>"]

tokio/README.md

@@ -56,7 +56,7 @@ Make sure you activated the full features of the tokio crate on Cargo.toml:
 
 ```toml
 [dependencies]
-tokio = { version = "1.43.2", features = ["full"] }
+tokio = { version = "1.43.3", features = ["full"] }
 ```
 Then, on your main.rs:
 

tokio/src/process/unix/pidfd_reaper.rs

@@ -95,14 +95,41 @@ where
     pidfd: PollEvented<Pidfd>,
 }
 
+fn display_eq(d: impl std::fmt::Display, s: &str) -> bool {
+    use std::fmt::Write;
+
+    struct FormatEq<'r> {
+        remainder: &'r str,
+        unequal: bool,
+    }
+
+    impl<'r> Write for FormatEq<'r> {
+        fn write_str(&mut self, s: &str) -> std::fmt::Result {
+            if !self.unequal {
+                if let Some(new_remainder) = self.remainder.strip_prefix(s) {
+                    self.remainder = new_remainder;
+                } else {
+                    self.unequal = true;
+                }
+            }
+            Ok(())
+        }
+    }
+
+    let mut fmt_eq = FormatEq {
+        remainder: s,
+        unequal: false,
+    };
+    let _ = write!(fmt_eq, "{d}");
+    fmt_eq.remainder.is_empty() && !fmt_eq.unequal
+}
+
 #[allow(deprecated)]
 fn is_rt_shutdown_err(err: &io::Error) -> bool {
     if let Some(inner) = err.get_ref() {
-        // Using `Error::description()` is more efficient than `format!("{inner}")`,
-        // so we use it here even if it is deprecated.
         err.kind() == io::ErrorKind::Other
             && inner.source().is_none()
-            && inner.description() == RUNTIME_SHUTTING_DOWN_ERROR
+            && display_eq(inner, RUNTIME_SHUTTING_DOWN_ERROR)
     } else {
         false
     }

tokio/src/runtime/task/state.rs

@@ -252,9 +252,15 @@ impl State {
     /// Transitions the state to `NOTIFIED`.
     pub(super) fn transition_to_notified_by_ref(&self) -> TransitionToNotifiedByRef {
         self.fetch_update_action(|mut snapshot| {
-            if snapshot.is_complete() || snapshot.is_notified() {
-                // There is nothing to do in this case.
+            if snapshot.is_complete() {
+                // The complete state is final
                 (TransitionToNotifiedByRef::DoNothing, None)
+            } else if snapshot.is_notified() {
+                // Even hough we have nothing to do in this branch,
+                // wake_by_ref() should synchronize-with the task starting execution,
+                // therefore we must use an Release store (with the same value),
+                // to pair with the Acquire in transition_to_running.
+                (TransitionToNotifiedByRef::DoNothing, Some(snapshot))
             } else if snapshot.is_running() {
                 // If the task is running, we mark it as notified, but we should
                 // not submit as the thread currently running the future is

tokio/src/sync/broadcast.rs

@@ -522,7 +522,7 @@ impl<T> Sender<T> {
             tail: Mutex::new(Tail {
                 pos: 0,
                 rx_cnt: receiver_count,
-                closed: false,
+                closed: receiver_count == 0,
                 waiters: LinkedList::new(),
             }),
             num_tx: AtomicUsize::new(1),

tokio/tests/sync_broadcast.rs

@@ -656,3 +656,15 @@ async fn receiver_recv_is_cooperative() {
         _ = tokio::task::yield_now() => {},
     }
 }
+
+#[tokio::test]
+async fn broadcast_sender_new_must_be_closed() {
+    let capacity = 1;
+    let tx: broadcast::Sender<()> = broadcast::Sender::new(capacity);
+
+    tx.send(()).unwrap_err();
+
+    let _rx = tx.subscribe();
+
+    tx.send(()).unwrap();
+}