diff --git a/Cargo.lock b/Cargo.lock
index ce71e2bf98950..32cb936bdaae3 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1825,13 +1825,13 @@ dependencies = [
 
 [[package]]
 name = "comfy-table"
-version = "7.1.0"
+version = "7.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7c64043d6c7b7a4c58e39e7efccfdea7b93d885a795d0c054a69dbbf4dd52686"
+checksum = "9ab77dbd8adecaf3f0db40581631b995f312a8a5ae3aa9993188bb8f23d83a5b"
 dependencies = [
- "crossterm 0.27.0",
- "strum 0.25.0",
- "strum_macros 0.25.2",
+ "crossterm 0.26.1",
+ "strum 0.24.1",
+ "strum_macros 0.24.3",
  "unicode-width",
 ]
 
@@ -2140,14 +2140,17 @@ dependencies = [
 
 [[package]]
 name = "crossterm"
-version = "0.27.0"
+version = "0.26.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f476fe445d41c9e991fd07515a6f463074b782242ccf4a5b7b1d1012e70824df"
+checksum = "a84cda67535339806297f1b331d6dd6320470d2a0fe65381e79ee9e156dd3d13"
 dependencies = [
- "bitflags 2.4.0",
+ "bitflags 1.3.2",
  "crossterm_winapi",
  "libc",
+ "mio",
  "parking_lot 0.12.1",
+ "signal-hook",
+ "signal-hook-mio",
  "winapi",
 ]
 
@@ -2445,10 +2448,11 @@ dependencies = [
 
 [[package]]
 name = "deranged"
-version = "0.3.8"
+version = "0.3.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f2696e8a945f658fd14dc3b87242e6b80cd0f36ff04ea560fa39082368847946"
+checksum = "0f32d04922c60427da6f9fef14d042d9edddef64cb9d4ce0d64d0685fbeb1fd3"
 dependencies = [
+ "powerfmt",
  "serde",
 ]
 
@@ -2551,8 +2555,7 @@ checksum = "86e3bdc80eee6e16b2b6b0f87fbc98c04bee3455e35174c0de1a125d0688c632"
 [[package]]
 name = "dlv-list"
 version = "0.5.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8aead04dc46b5f263c25721cf25c9e595951d15055f8063f92392fa0d7f64cf4"
+source = "git+https://github.com/sgodwincs/dlv-list-rs.git?rev=5bbc5d0#5bbc5d0cc84f257e173d851f8dc1674fb6e46f95"
 dependencies = [
  "const-random",
 ]
@@ -3015,7 +3018,7 @@ dependencies = [
 [[package]]
 name = "foyer"
 version = "0.1.0"
-source = "git+https://github.com/mrcroxx/foyer?rev=5d0134b#5d0134b28c0edb03277b01ce08b035ef52c1b783"
+source = "git+https://github.com/MrCroxx/foyer?rev=2261151#2261151107ad362851f5fff9ce4fa56e61911b10"
 dependencies = [
  "foyer-common",
  "foyer-intrusive",
@@ -3026,7 +3029,7 @@ dependencies = [
 [[package]]
 name = "foyer-common"
 version = "0.1.0"
-source = "git+https://github.com/mrcroxx/foyer?rev=5d0134b#5d0134b28c0edb03277b01ce08b035ef52c1b783"
+source = "git+https://github.com/MrCroxx/foyer?rev=2261151#2261151107ad362851f5fff9ce4fa56e61911b10"
 dependencies = [
  "bytes",
  "foyer-workspace-hack",
@@ -3041,7 +3044,7 @@ dependencies = [
 [[package]]
 name = "foyer-intrusive"
 version = "0.1.0"
-source = "git+https://github.com/mrcroxx/foyer?rev=5d0134b#5d0134b28c0edb03277b01ce08b035ef52c1b783"
+source = "git+https://github.com/MrCroxx/foyer?rev=2261151#2261151107ad362851f5fff9ce4fa56e61911b10"
 dependencies = [
  "bytes",
  "cmsketch",
@@ -3058,7 +3061,7 @@ dependencies = [
 [[package]]
 name = "foyer-storage"
 version = "0.1.0"
-source = "git+https://github.com/mrcroxx/foyer?rev=5d0134b#5d0134b28c0edb03277b01ce08b035ef52c1b783"
+source = "git+https://github.com/MrCroxx/foyer?rev=2261151#2261151107ad362851f5fff9ce4fa56e61911b10"
 dependencies = [
  "anyhow",
  "async-channel",
@@ -3087,7 +3090,7 @@ dependencies = [
 [[package]]
 name = "foyer-workspace-hack"
 version = "0.1.0"
-source = "git+https://github.com/mrcroxx/foyer?rev=5d0134b#5d0134b28c0edb03277b01ce08b035ef52c1b783"
+source = "git+https://github.com/MrCroxx/foyer?rev=2261151#2261151107ad362851f5fff9ce4fa56e61911b10"
 dependencies = [
  "crossbeam-utils",
  "either",
@@ -3225,9 +3228,9 @@ dependencies = [
 
 [[package]]
 name = "futures-async-stream"
-version = "0.2.7"
+version = "0.2.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f529ccdeacfa2446a9577041686cf1abb839b1b3e15fee4c1b1232ab3b7d799f"
+checksum = "379790776b0d953337df4ab7ecc51936c66ea112484cad7912907b1d34253ebf"
 dependencies = [
  "futures-async-stream-macro",
  "futures-core",
@@ -3236,13 +3239,13 @@ dependencies = [
 
 [[package]]
 name = "futures-async-stream-macro"
-version = "0.2.7"
+version = "0.2.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ca2b48ee06dc8d2808ba5ebad075d06c3406085bb19deaac33be64c39113bf80"
+checksum = "5df2c13d48c8cb8a3ec093ede6f0f4482f327d7bb781120c5fb483ef0f17e758"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 1.0.109",
+ "syn 2.0.37",
 ]
 
 [[package]]
@@ -5492,8 +5495,7 @@ dependencies = [
 [[package]]
 name = "ordered-multimap"
 version = "0.6.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4ed8acf08e98e744e5384c8bc63ceb0364e68a6854187221c18df61c4797690e"
+source = "git+https://github.com/risingwavelabs/ordered-multimap-rs.git?rev=19c743f#19c743f3e3d106c99ba37628f06a2ca6faa2284f"
 dependencies = [
  "dlv-list",
  "hashbrown 0.13.2",
@@ -6023,6 +6025,12 @@ dependencies = [
  "serde_json",
 ]
 
+[[package]]
+name = "powerfmt"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "439ee305def115ba05938db6eb1644ff94165c5ab5e9420d1c1bcedbba909391"
+
 [[package]]
 name = "pprof"
 version = "0.13.0"
@@ -9803,14 +9811,15 @@ dependencies = [
 
 [[package]]
 name = "time"
-version = "0.3.28"
+version = "0.3.30"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "17f6bb557fd245c28e6411aa56b6403c689ad95061f50e4be16c274e70a17e48"
+checksum = "c4a34ab300f2dee6e562c10a046fc05e358b29f9bf92277f30c3c8d82275f6f5"
 dependencies = [
  "deranged",
  "itoa",
  "libc",
  "num_threads",
+ "powerfmt",
  "serde",
  "time-core",
  "time-macros",
@@ -9818,15 +9827,15 @@ dependencies = [
 
 [[package]]
 name = "time-core"
-version = "0.1.1"
+version = "0.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7300fbefb4dadc1af235a9cef3737cea692a9d97e1b9cbcd4ebdae6f8868e6fb"
+checksum = "ef927ca75afb808a4d64dd374f00a2adf8d0fcff8e7b184af886c3c87ec4a3f3"
 
 [[package]]
 name = "time-macros"
-version = "0.2.14"
+version = "0.2.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1a942f44339478ef67935ab2bbaec2fb0322496cf3cbe84b261e06ac3814c572"
+checksum = "4ad70d68dba9e1f8aceda7aa6711965dfec1cac869f311a51bd08b3a2ccbce20"
 dependencies = [
  "time-core",
 ]
diff --git a/Cargo.toml b/Cargo.toml
index ef09221b818a2..f8a9b7d0e2fa5 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -97,7 +97,7 @@ aws-smithy-types = "0.55"
 aws-endpoint = "0.55"
 aws-types = "0.55"
 etcd-client = { package = "madsim-etcd-client", version = "0.4" }
-futures-async-stream = "0.2"
+futures-async-stream = "0.2.9"
 hytra = "0.1"
 rdkafka = { package = "madsim-rdkafka", version = "0.3.0", features = [
   "cmake-build",
@@ -165,6 +165,8 @@ unused_must_use = "forbid"
 future_incompatible = "warn"
 nonstandard_style = "warn"
 rust_2018_idioms = "warn"
+# Backward compatibility is not important for an application.
+async_fn_in_trait = "allow"
 
 [workspace.lints.clippy]
 uninlined_format_args = "allow"
@@ -229,8 +231,8 @@ opt-level = 2
 incremental = false
 debug = 1
 
-# Patch third-party crates for deterministic simulation.
 [patch.crates-io]
+# Patch third-party crates for deterministic simulation.
 quanta = { git = "https://github.com/madsim-rs/quanta.git", rev = "948bdc3" }
 getrandom = { git = "https://github.com/madsim-rs/getrandom.git", rev = "8daf97e" }
 tokio-stream = { git = "https://github.com/madsim-rs/tokio.git", rev = "fe39bb8e" }
@@ -238,3 +240,8 @@ tokio-retry = { git = "https://github.com/madsim-rs/rust-tokio-retry.git", rev =
 tokio-postgres = { git = "https://github.com/madsim-rs/rust-postgres.git", rev = "ac00d88" }
 # patch: unlimit 4MB message size for grpc client
 etcd-client = { git = "https://github.com/risingwavelabs/etcd-client.git", rev = "4e84d40" }
+
+# Patch for coverage_attribute.
+# https://github.com/sgodwincs/dlv-list-rs/pull/19#issuecomment-1774786289
+dlv-list = { git = "https://github.com/sgodwincs/dlv-list-rs.git", rev = "5bbc5d0" }
+ordered-multimap = { git = "https://github.com/risingwavelabs/ordered-multimap-rs.git", rev = "19c743f" }
diff --git a/ci/build-ci-image.sh b/ci/build-ci-image.sh
index 43ff81ade2b85..59c88e5e9a9ae 100755
--- a/ci/build-ci-image.sh
+++ b/ci/build-ci-image.sh
@@ -13,7 +13,7 @@ cat ../rust-toolchain
 # !!! CHANGE THIS WHEN YOU WANT TO BUMP CI IMAGE !!! #
 #          AND ALSO docker-compose.yml               #
 ######################################################
-export BUILD_ENV_VERSION=v20230919
+export BUILD_ENV_VERSION=v20231022
 
 export BUILD_TAG="public.ecr.aws/x5u3w5h6/rw-build-env:${BUILD_ENV_VERSION}"
 
diff --git a/ci/docker-compose.yml b/ci/docker-compose.yml
index 6fe7cfbfdeca2..48d88e7096889 100644
--- a/ci/docker-compose.yml
+++ b/ci/docker-compose.yml
@@ -71,7 +71,7 @@ services:
       retries: 5
 
   source-test-env:
-    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230919
+    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20231022
     depends_on:
       - mysql
       - db
@@ -81,7 +81,7 @@ services:
       - ..:/risingwave
 
   sink-test-env:
-    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230919
+    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20231022
     depends_on:
       - mysql
       - db
@@ -92,12 +92,12 @@ services:
       - ..:/risingwave
 
   rw-build-env:
-    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230919
+    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20231022
     volumes:
       - ..:/risingwave
 
   ci-flamegraph-env:
-    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230919
+    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20231022
     # NOTE(kwannoel): This is used in order to permit
     # syscalls for `nperf` (perf_event_open),
     # so it can do CPU profiling.
@@ -108,7 +108,7 @@ services:
       - ..:/risingwave
 
   regress-test-env:
-    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230919
+    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20231022
     depends_on:
       db:
         condition: service_healthy
diff --git a/ci/rust-toolchain b/ci/rust-toolchain
index ebc0b6c285a4e..fe2a026f6e40f 100644
--- a/ci/rust-toolchain
+++ b/ci/rust-toolchain
@@ -1,2 +1,2 @@
 [toolchain]
-channel = "nightly-2023-09-09"
+channel = "nightly-2023-10-21"
diff --git a/src/batch/src/lib.rs b/src/batch/src/lib.rs
index 9104c96c951f5..809c096eb49df 100644
--- a/src/batch/src/lib.rs
+++ b/src/batch/src/lib.rs
@@ -17,8 +17,8 @@
 #![feature(trait_alias)]
 #![feature(exact_size_is_empty)]
 #![feature(type_alias_impl_trait)]
-#![cfg_attr(coverage, feature(no_coverage))]
-#![feature(generators)]
+#![cfg_attr(coverage, feature(coverage_attribute))]
+#![feature(coroutines)]
 #![feature(proc_macro_hygiene, stmt_expr_attributes)]
 #![feature(iterator_try_collect)]
 #![feature(lint_reasons)]
@@ -27,13 +27,11 @@
 #![feature(let_chains)]
 #![feature(bound_map)]
 #![feature(int_roundings)]
-#![feature(async_fn_in_trait)]
 #![feature(allocator_api)]
 #![feature(impl_trait_in_assoc_type)]
 #![feature(result_option_inspect)]
 #![feature(assert_matches)]
 #![feature(lazy_cell)]
-#![feature(return_position_impl_trait_in_trait)]
 
 mod error;
 pub mod exchange_source;
diff --git a/src/batch/src/rpc/service/task_service.rs b/src/batch/src/rpc/service/task_service.rs
index b49a023acb22b..fb60e352ec293 100644
--- a/src/batch/src/rpc/service/task_service.rs
+++ b/src/batch/src/rpc/service/task_service.rs
@@ -53,7 +53,7 @@ impl TaskService for BatchServiceImpl {
     type CreateTaskStream = ReceiverStream<TaskInfoResponseResult>;
     type ExecuteStream = ReceiverStream<GetDataResponseResult>;
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn create_task(
         &self,
         request: Request<CreateTaskRequest>,
@@ -97,7 +97,7 @@ impl TaskService for BatchServiceImpl {
         }
     }
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn cancel_task(
         &self,
         req: Request<CancelTaskRequest>,
@@ -109,7 +109,7 @@ impl TaskService for BatchServiceImpl {
         Ok(Response::new(CancelTaskResponse { status: None }))
     }
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn execute(
         &self,
         req: Request<ExecuteRequest>,
diff --git a/src/cmd/src/bin/compactor.rs b/src/cmd/src/bin/compactor.rs
index 21b7db2405e2d..554168d8a6683 100644
--- a/src/cmd/src/bin/compactor.rs
+++ b/src/cmd/src/bin/compactor.rs
@@ -12,6 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![cfg_attr(coverage, feature(no_coverage))]
+#![cfg_attr(coverage, feature(coverage_attribute))]
 
 risingwave_cmd::main!(compactor);
diff --git a/src/cmd/src/bin/compute_node.rs b/src/cmd/src/bin/compute_node.rs
index 0bb1e5211ac57..a24d132b70b94 100644
--- a/src/cmd/src/bin/compute_node.rs
+++ b/src/cmd/src/bin/compute_node.rs
@@ -12,6 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![cfg_attr(coverage, feature(no_coverage))]
+#![cfg_attr(coverage, feature(coverage_attribute))]
 
 risingwave_cmd::main!(compute);
diff --git a/src/cmd/src/bin/ctl.rs b/src/cmd/src/bin/ctl.rs
index 38345c7a3fc2e..7b4c3132e747d 100644
--- a/src/cmd/src/bin/ctl.rs
+++ b/src/cmd/src/bin/ctl.rs
@@ -12,6 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![cfg_attr(coverage, feature(no_coverage))]
+#![cfg_attr(coverage, feature(coverage_attribute))]
 
 risingwave_cmd::main!(ctl);
diff --git a/src/cmd/src/bin/frontend_node.rs b/src/cmd/src/bin/frontend_node.rs
index 32d563be109fc..546bacbf1a901 100644
--- a/src/cmd/src/bin/frontend_node.rs
+++ b/src/cmd/src/bin/frontend_node.rs
@@ -12,6 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![cfg_attr(coverage, feature(no_coverage))]
+#![cfg_attr(coverage, feature(coverage_attribute))]
 
 risingwave_cmd::main!(frontend);
diff --git a/src/cmd/src/bin/meta_node.rs b/src/cmd/src/bin/meta_node.rs
index 032cc6bc28285..4bebfc5f915a2 100644
--- a/src/cmd/src/bin/meta_node.rs
+++ b/src/cmd/src/bin/meta_node.rs
@@ -12,6 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![cfg_attr(coverage, feature(no_coverage))]
+#![cfg_attr(coverage, feature(coverage_attribute))]
 
 risingwave_cmd::main!(meta);
diff --git a/src/cmd/src/lib.rs b/src/cmd/src/lib.rs
index 12de26657bd33..93df94a63816a 100644
--- a/src/cmd/src/lib.rs
+++ b/src/cmd/src/lib.rs
@@ -30,7 +30,7 @@ macro_rules! main {
         #[cfg(not(enable_task_local_alloc))]
         risingwave_common::enable_jemalloc!();
 
-        #[cfg_attr(coverage, no_coverage)]
+        #[cfg_attr(coverage, coverage(off))]
         fn main() {
             let opts = clap::Parser::parse();
             $crate::$component(opts);
diff --git a/src/cmd_all/src/bin/risingwave.rs b/src/cmd_all/src/bin/risingwave.rs
index 3e9088e16b9e2..b7693c6fa06a2 100644
--- a/src/cmd_all/src/bin/risingwave.rs
+++ b/src/cmd_all/src/bin/risingwave.rs
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![cfg_attr(coverage, feature(no_coverage))]
+#![cfg_attr(coverage, feature(coverage_attribute))]
 
 use std::str::FromStr;
 
@@ -158,7 +158,7 @@ impl Component {
     }
 }
 
-#[cfg_attr(coverage, no_coverage)]
+#[cfg_attr(coverage, coverage(off))]
 fn main() -> Result<()> {
     let risingwave = || {
         command!(BINARY_NAME)
diff --git a/src/common/proc_macro/src/config.rs b/src/common/proc_macro/src/config.rs
index 285834eb123cf..6e369fbad33eb 100644
--- a/src/common/proc_macro/src/config.rs
+++ b/src/common/proc_macro/src/config.rs
@@ -41,7 +41,7 @@ fn type_is_option(ty: &syn::Type) -> bool {
     false
 }
 
-#[cfg_attr(coverage, no_coverage)]
+#[cfg_attr(coverage, coverage(off))]
 pub fn produce_override_config(input: DeriveInput) -> TokenStream {
     let syn::Data::Struct(syn::DataStruct { fields, .. }) = input.data else {
         abort!(input, "Only struct is supported");
diff --git a/src/common/proc_macro/src/lib.rs b/src/common/proc_macro/src/lib.rs
index 060ee1950624e..a11e407c6c053 100644
--- a/src/common/proc_macro/src/lib.rs
+++ b/src/common/proc_macro/src/lib.rs
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![cfg_attr(coverage, feature(no_coverage))]
+#![cfg_attr(coverage, feature(coverage_attribute))]
 
 use estimate_size::{
     add_trait_bounds, extract_ignored_generics_list, has_nested_flag_attribute_list,
@@ -52,7 +52,7 @@ mod estimate_size;
 ///     }
 /// }
 /// ```
-#[cfg_attr(coverage, no_coverage)]
+#[cfg_attr(coverage, coverage(off))]
 #[proc_macro_derive(OverrideConfig, attributes(override_opts))]
 #[proc_macro_error]
 pub fn override_config(input: TokenStream) -> TokenStream {
diff --git a/src/common/src/lib.rs b/src/common/src/lib.rs
index 2a3575d8dae78..fbcd3854fa572 100644
--- a/src/common/src/lib.rs
+++ b/src/common/src/lib.rs
@@ -24,12 +24,11 @@
 #![feature(trusted_len)]
 #![feature(allocator_api)]
 #![feature(lint_reasons)]
-#![feature(generators)]
+#![feature(coroutines)]
 #![feature(map_try_insert)]
 #![feature(lazy_cell)]
 #![feature(error_generic_member_access)]
 #![feature(let_chains)]
-#![feature(return_position_impl_trait_in_trait)]
 #![feature(portable_simd)]
 #![feature(array_chunks)]
 #![feature(inline_const_pat)]
@@ -43,7 +42,6 @@
 #![feature(result_option_inspect)]
 #![feature(map_entry_replace)]
 #![feature(negative_impls)]
-#![feature(async_fn_in_trait)]
 #![feature(bound_map)]
 #![feature(array_methods)]
 
diff --git a/src/common/src/types/ordered.rs b/src/common/src/types/ordered.rs
index 75b07e529d7b9..68cd6329287e2 100644
--- a/src/common/src/types/ordered.rs
+++ b/src/common/src/types/ordered.rs
@@ -138,7 +138,7 @@ impl<T: DefaultOrd> From<T> for DefaultOrdered<T> {
     }
 }
 
-#[allow(clippy::incorrect_partial_ord_impl_on_ord_type)]
+#[allow(clippy::non_canonical_partial_ord_impl)]
 impl<T: DefaultOrd> PartialOrd for DefaultOrdered<T> {
     fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
         self.0.default_partial_cmp(other.as_inner())
diff --git a/src/compute/src/lib.rs b/src/compute/src/lib.rs
index 65bf59eedf19e..fc5ae9ff19854 100644
--- a/src/compute/src/lib.rs
+++ b/src/compute/src/lib.rs
@@ -13,14 +13,14 @@
 // limitations under the License.
 
 #![feature(trait_alias)]
-#![feature(generators)]
+#![feature(coroutines)]
 #![feature(type_alias_impl_trait)]
 #![feature(let_chains)]
 #![feature(result_option_inspect)]
 #![feature(lint_reasons)]
 #![feature(impl_trait_in_assoc_type)]
 #![feature(lazy_cell)]
-#![cfg_attr(coverage, feature(no_coverage))]
+#![cfg_attr(coverage, feature(coverage_attribute))]
 
 #[macro_use]
 extern crate tracing;
diff --git a/src/compute/src/rpc/service/exchange_service.rs b/src/compute/src/rpc/service/exchange_service.rs
index b59cc39587c2f..6225cef2a7e30 100644
--- a/src/compute/src/rpc/service/exchange_service.rs
+++ b/src/compute/src/rpc/service/exchange_service.rs
@@ -49,7 +49,7 @@ impl ExchangeService for ExchangeServiceImpl {
     type GetDataStream = BatchDataStream;
     type GetStreamStream = StreamDataStream;
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn get_data(
         &self,
         request: Request<GetDataRequest>,
diff --git a/src/compute/src/rpc/service/monitor_service.rs b/src/compute/src/rpc/service/monitor_service.rs
index 97a0b80773791..8fc24664ec016 100644
--- a/src/compute/src/rpc/service/monitor_service.rs
+++ b/src/compute/src/rpc/service/monitor_service.rs
@@ -53,7 +53,7 @@ impl MonitorServiceImpl {
 
 #[async_trait::async_trait]
 impl MonitorService for MonitorServiceImpl {
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn stack_trace(
         &self,
         request: Request<StackTraceRequest>,
@@ -85,7 +85,7 @@ impl MonitorService for MonitorServiceImpl {
         }))
     }
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn profiling(
         &self,
         request: Request<ProfilingRequest>,
@@ -115,7 +115,7 @@ impl MonitorService for MonitorServiceImpl {
         }
     }
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn heap_profiling(
         &self,
         request: Request<HeapProfilingRequest>,
@@ -166,7 +166,7 @@ impl MonitorService for MonitorServiceImpl {
         }
     }
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn list_heap_profiling(
         &self,
         _request: Request<ListHeapProfilingRequest>,
@@ -206,7 +206,7 @@ impl MonitorService for MonitorServiceImpl {
         }))
     }
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn analyze_heap(
         &self,
         request: Request<AnalyzeHeapRequest>,
diff --git a/src/compute/src/rpc/service/stream_service.rs b/src/compute/src/rpc/service/stream_service.rs
index 525364b60dc1c..1c1448b3d1e45 100644
--- a/src/compute/src/rpc/service/stream_service.rs
+++ b/src/compute/src/rpc/service/stream_service.rs
@@ -45,7 +45,7 @@ impl StreamServiceImpl {
 
 #[async_trait::async_trait]
 impl StreamService for StreamServiceImpl {
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn update_actors(
         &self,
         request: Request<UpdateActorsRequest>,
@@ -61,7 +61,7 @@ impl StreamService for StreamServiceImpl {
         }
     }
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn build_actors(
         &self,
         request: Request<BuildActorsRequest>,
@@ -85,7 +85,7 @@ impl StreamService for StreamServiceImpl {
         }
     }
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn broadcast_actor_info_table(
         &self,
         request: Request<BroadcastActorInfoTableRequest>,
@@ -104,7 +104,7 @@ impl StreamService for StreamServiceImpl {
         }
     }
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn drop_actors(
         &self,
         request: Request<DropActorsRequest>,
@@ -118,7 +118,7 @@ impl StreamService for StreamServiceImpl {
         }))
     }
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn force_stop_actors(
         &self,
         request: Request<ForceStopActorsRequest>,
@@ -132,7 +132,7 @@ impl StreamService for StreamServiceImpl {
         }))
     }
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn inject_barrier(
         &self,
         request: Request<InjectBarrierRequest>,
@@ -173,7 +173,7 @@ impl StreamService for StreamServiceImpl {
         }))
     }
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn barrier_complete(
         &self,
         request: Request<BarrierCompleteRequest>,
@@ -243,7 +243,7 @@ impl StreamService for StreamServiceImpl {
         }))
     }
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn wait_epoch_commit(
         &self,
         request: Request<WaitEpochCommitRequest>,
diff --git a/src/compute/tests/cdc_tests.rs b/src/compute/tests/cdc_tests.rs
index b3e39ece95002..6a50b8410bbd4 100644
--- a/src/compute/tests/cdc_tests.rs
+++ b/src/compute/tests/cdc_tests.rs
@@ -13,7 +13,7 @@
 // limitations under the License.
 
 #![feature(let_chains)]
-#![feature(generators)]
+#![feature(coroutines)]
 
 use std::sync::atomic::AtomicU64;
 use std::sync::Arc;
diff --git a/src/compute/tests/integration_tests.rs b/src/compute/tests/integration_tests.rs
index a43ae2e5762da..6d7e93365c275 100644
--- a/src/compute/tests/integration_tests.rs
+++ b/src/compute/tests/integration_tests.rs
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![feature(generators)]
+#![feature(coroutines)]
 #![feature(proc_macro_hygiene, stmt_expr_attributes)]
 
 use std::sync::atomic::AtomicU64;
diff --git a/src/connector/Cargo.toml b/src/connector/Cargo.toml
index 4886b1b52fcc5..87d2a0bdef689 100644
--- a/src/connector/Cargo.toml
+++ b/src/connector/Cargo.toml
@@ -113,7 +113,7 @@ strum = "0.25"
 strum_macros = "0.25"
 tempfile = "3"
 thiserror = "1"
-time = "0.3.28"
+time = "0.3.30"
 tokio = { version = "0.2", package = "madsim-tokio", features = [
     "rt",
     "rt-multi-thread",
diff --git a/src/connector/src/lib.rs b/src/connector/src/lib.rs
index 8ccf62486ce65..aa613b4043c23 100644
--- a/src/connector/src/lib.rs
+++ b/src/connector/src/lib.rs
@@ -14,7 +14,7 @@
 
 #![expect(dead_code)]
 #![allow(clippy::derive_partial_eq_without_eq)]
-#![feature(generators)]
+#![feature(coroutines)]
 #![feature(proc_macro_hygiene)]
 #![feature(stmt_expr_attributes)]
 #![feature(box_patterns)]
@@ -25,11 +25,9 @@
 #![feature(let_chains)]
 #![feature(box_into_inner)]
 #![feature(type_alias_impl_trait)]
-#![feature(return_position_impl_trait_in_trait)]
-#![feature(async_fn_in_trait)]
 #![feature(associated_type_defaults)]
 #![feature(impl_trait_in_assoc_type)]
-#![feature(iter_from_generator)]
+#![feature(iter_from_coroutine)]
 #![feature(if_let_guard)]
 #![feature(iterator_try_collect)]
 
diff --git a/src/connector/src/sink/formatter/append_only.rs b/src/connector/src/sink/formatter/append_only.rs
index 523a52dab91bb..f0efcc21d9009 100644
--- a/src/connector/src/sink/formatter/append_only.rs
+++ b/src/connector/src/sink/formatter/append_only.rs
@@ -40,7 +40,7 @@ impl<KE: RowEncoder, VE: RowEncoder> SinkFormatter for AppendOnlyFormatter<KE, V
         &self,
         chunk: &StreamChunk,
     ) -> impl Iterator<Item = Result<(Option<Self::K>, Option<Self::V>)>> {
-        std::iter::from_generator(|| {
+        std::iter::from_coroutine(|| {
             for (op, row) in chunk.rows() {
                 if op != Op::Insert {
                     continue;
diff --git a/src/connector/src/sink/formatter/debezium_json.rs b/src/connector/src/sink/formatter/debezium_json.rs
index 637aa23f06410..ce98daab88756 100644
--- a/src/connector/src/sink/formatter/debezium_json.rs
+++ b/src/connector/src/sink/formatter/debezium_json.rs
@@ -85,7 +85,7 @@ impl SinkFormatter for DebeziumJsonFormatter {
         &self,
         chunk: &StreamChunk,
     ) -> impl Iterator<Item = Result<(Option<Value>, Option<Value>)>> {
-        std::iter::from_generator(|| {
+        std::iter::from_coroutine(|| {
             let DebeziumJsonFormatter {
                 schema,
                 pk_indices,
diff --git a/src/connector/src/sink/formatter/upsert.rs b/src/connector/src/sink/formatter/upsert.rs
index 6ef2b5f2ca333..af8e70ff92850 100644
--- a/src/connector/src/sink/formatter/upsert.rs
+++ b/src/connector/src/sink/formatter/upsert.rs
@@ -40,7 +40,7 @@ impl<KE: RowEncoder, VE: RowEncoder> SinkFormatter for UpsertFormatter<KE, VE> {
         &self,
         chunk: &StreamChunk,
     ) -> impl Iterator<Item = Result<(Option<Self::K>, Option<Self::V>)>> {
-        std::iter::from_generator(|| {
+        std::iter::from_coroutine(|| {
             for (op, row) in chunk.rows() {
                 let event_key_object = Some(tri!(self.key_encoder.encode(row)));
 
diff --git a/src/expr/core/src/lib.rs b/src/expr/core/src/lib.rs
index c2f46d5632274..b49c4ae161dfc 100644
--- a/src/expr/core/src/lib.rs
+++ b/src/expr/core/src/lib.rs
@@ -17,7 +17,7 @@
 #![feature(lint_reasons)]
 #![feature(iterator_try_collect)]
 #![feature(lazy_cell)]
-#![feature(generators)]
+#![feature(coroutines)]
 #![feature(arc_unwrap_or_clone)]
 #![feature(never_type)]
 
diff --git a/src/expr/impl/src/lib.rs b/src/expr/impl/src/lib.rs
index a5906e4320282..6ea82d30ac5f1 100644
--- a/src/expr/impl/src/lib.rs
+++ b/src/expr/impl/src/lib.rs
@@ -28,7 +28,7 @@
 #![feature(exclusive_range_pattern)]
 #![feature(lazy_cell)]
 #![feature(round_ties_even)]
-#![feature(generators)]
+#![feature(coroutines)]
 #![feature(test)]
 #![feature(arc_unwrap_or_clone)]
 
diff --git a/src/frontend/src/binder/mod.rs b/src/frontend/src/binder/mod.rs
index 974730cd16237..f1038f9bf5943 100644
--- a/src/frontend/src/binder/mod.rs
+++ b/src/frontend/src/binder/mod.rs
@@ -363,6 +363,13 @@ impl Binder {
     }
 }
 
+/// The column name stored in [`BindContext`] for a column without an alias.
+pub const UNNAMED_COLUMN: &str = "?column?";
+/// The table name stored in [`BindContext`] for a subquery without an alias.
+const UNNAMED_SUBQUERY: &str = "?subquery?";
+/// The table name stored in [`BindContext`] for a column group.
+const COLUMN_GROUP_PREFIX: &str = "?column_group_id?";
+
 #[cfg(test)]
 pub mod test_utils {
     use risingwave_common::types::DataType;
@@ -380,10 +387,3 @@ pub mod test_utils {
         Binder::new_with_param_types(&SessionImpl::mock(), param_types)
     }
 }
-
-/// The column name stored in [`BindContext`] for a column without an alias.
-pub const UNNAMED_COLUMN: &str = "?column?";
-/// The table name stored in [`BindContext`] for a subquery without an alias.
-const UNNAMED_SUBQUERY: &str = "?subquery?";
-/// The table name stored in [`BindContext`] for a column group.
-const COLUMN_GROUP_PREFIX: &str = "?column_group_id?";
diff --git a/src/frontend/src/lib.rs b/src/frontend/src/lib.rs
index 0a036b8e96233..450f49b6394cf 100644
--- a/src/frontend/src/lib.rs
+++ b/src/frontend/src/lib.rs
@@ -15,7 +15,7 @@
 #![allow(clippy::derive_partial_eq_without_eq)]
 #![feature(map_try_insert)]
 #![feature(negative_impls)]
-#![feature(generators)]
+#![feature(coroutines)]
 #![feature(proc_macro_hygiene, stmt_expr_attributes)]
 #![feature(trait_alias)]
 #![feature(extract_if)]
@@ -32,7 +32,6 @@
 #![feature(extend_one)]
 #![feature(type_alias_impl_trait)]
 #![feature(impl_trait_in_assoc_type)]
-#![feature(async_fn_in_trait)]
 #![feature(result_flattening)]
 #![recursion_limit = "256"]
 
diff --git a/src/meta/node/src/lib.rs b/src/meta/node/src/lib.rs
index 55c7b27b0c80a..bf1bddad2070f 100644
--- a/src/meta/node/src/lib.rs
+++ b/src/meta/node/src/lib.rs
@@ -14,7 +14,7 @@
 
 #![feature(lint_reasons)]
 #![feature(let_chains)]
-#![cfg_attr(coverage, feature(no_coverage))]
+#![cfg_attr(coverage, feature(coverage_attribute))]
 
 mod server;
 use std::time::Duration;
diff --git a/src/meta/service/src/ddl_service.rs b/src/meta/service/src/ddl_service.rs
index 935d398aeacb0..2fa5f50e15666 100644
--- a/src/meta/service/src/ddl_service.rs
+++ b/src/meta/service/src/ddl_service.rs
@@ -717,7 +717,7 @@ impl DdlService for DdlServiceImpl {
         }))
     }
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn get_tables(
         &self,
         request: Request<GetTablesRequest>,
diff --git a/src/meta/service/src/heartbeat_service.rs b/src/meta/service/src/heartbeat_service.rs
index 7c51b39346894..e31058ff2bdc5 100644
--- a/src/meta/service/src/heartbeat_service.rs
+++ b/src/meta/service/src/heartbeat_service.rs
@@ -32,7 +32,7 @@ impl HeartbeatServiceImpl {
 
 #[async_trait::async_trait]
 impl HeartbeatService for HeartbeatServiceImpl {
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn heartbeat(
         &self,
         request: Request<HeartbeatRequest>,
diff --git a/src/meta/service/src/lib.rs b/src/meta/service/src/lib.rs
index 0d473a6ed031f..6c8cc11f8971c 100644
--- a/src/meta/service/src/lib.rs
+++ b/src/meta/service/src/lib.rs
@@ -16,7 +16,7 @@
 #![feature(let_chains)]
 #![feature(lazy_cell)]
 #![feature(impl_trait_in_assoc_type)]
-#![cfg_attr(coverage, feature(no_coverage))]
+#![cfg_attr(coverage, feature(coverage_attribute))]
 
 use risingwave_meta::*;
 
diff --git a/src/meta/service/src/meta_member_service.rs b/src/meta/service/src/meta_member_service.rs
index 25c4c7ad4cc84..5753061176e8c 100644
--- a/src/meta/service/src/meta_member_service.rs
+++ b/src/meta/service/src/meta_member_service.rs
@@ -36,7 +36,7 @@ impl MetaMemberServiceImpl {
 
 #[async_trait::async_trait]
 impl MetaMemberService for MetaMemberServiceImpl {
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn members(
         &self,
         _request: Request<MembersRequest>,
diff --git a/src/meta/service/src/notification_service.rs b/src/meta/service/src/notification_service.rs
index bd247c1e18980..0fcc470a70e39 100644
--- a/src/meta/service/src/notification_service.rs
+++ b/src/meta/service/src/notification_service.rs
@@ -207,7 +207,7 @@ impl NotificationServiceImpl {
 impl NotificationService for NotificationServiceImpl {
     type SubscribeStream = UnboundedReceiverStream<Notification>;
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn subscribe(
         &self,
         request: Request<SubscribeRequest>,
diff --git a/src/meta/service/src/scale_service.rs b/src/meta/service/src/scale_service.rs
index f231ea5f4955d..676180adc7581 100644
--- a/src/meta/service/src/scale_service.rs
+++ b/src/meta/service/src/scale_service.rs
@@ -59,7 +59,7 @@ impl ScaleServiceImpl {
 
 #[async_trait::async_trait]
 impl ScaleService for ScaleServiceImpl {
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn get_cluster_info(
         &self,
         _: Request<GetClusterInfoRequest>,
@@ -110,7 +110,7 @@ impl ScaleService for ScaleServiceImpl {
         }))
     }
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn reschedule(
         &self,
         request: Request<RescheduleRequest>,
@@ -174,7 +174,7 @@ impl ScaleService for ScaleServiceImpl {
         }))
     }
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn get_reschedule_plan(
         &self,
         request: Request<GetReschedulePlanRequest>,
diff --git a/src/meta/service/src/stream_service.rs b/src/meta/service/src/stream_service.rs
index ef232d9b04ffd..92af1d4beb707 100644
--- a/src/meta/service/src/stream_service.rs
+++ b/src/meta/service/src/stream_service.rs
@@ -59,7 +59,7 @@ impl StreamServiceImpl {
 
 #[async_trait::async_trait]
 impl StreamManagerService for StreamServiceImpl {
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn flush(&self, request: Request<FlushRequest>) -> TonicResponse<FlushResponse> {
         self.env.idle_manager().record_activity();
         let req = request.into_inner();
@@ -71,7 +71,7 @@ impl StreamManagerService for StreamServiceImpl {
         }))
     }
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn pause(&self, _: Request<PauseRequest>) -> Result<Response<PauseResponse>, Status> {
         let i = self
             .barrier_scheduler
@@ -83,7 +83,7 @@ impl StreamManagerService for StreamServiceImpl {
         }))
     }
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn resume(&self, _: Request<ResumeRequest>) -> Result<Response<ResumeResponse>, Status> {
         let i = self
             .barrier_scheduler
@@ -122,7 +122,7 @@ impl StreamManagerService for StreamServiceImpl {
         }))
     }
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn list_table_fragments(
         &self,
         request: Request<ListTableFragmentsRequest>,
@@ -165,7 +165,7 @@ impl StreamManagerService for StreamServiceImpl {
         }))
     }
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn list_table_fragment_states(
         &self,
         _request: Request<ListTableFragmentStatesRequest>,
@@ -186,7 +186,7 @@ impl StreamManagerService for StreamServiceImpl {
         }))
     }
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn list_fragment_distribution(
         &self,
         _request: Request<ListFragmentDistributionRequest>,
@@ -215,7 +215,7 @@ impl StreamManagerService for StreamServiceImpl {
         }))
     }
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn list_actor_states(
         &self,
         _request: Request<ListActorStatesRequest>,
diff --git a/src/meta/service/src/user_service.rs b/src/meta/service/src/user_service.rs
index 8c982521b112a..cb290766e6fd1 100644
--- a/src/meta/service/src/user_service.rs
+++ b/src/meta/service/src/user_service.rs
@@ -107,7 +107,7 @@ impl UserServiceImpl {
 
 #[async_trait::async_trait]
 impl UserService for UserServiceImpl {
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn create_user(
         &self,
         request: Request<CreateUserRequest>,
@@ -128,7 +128,7 @@ impl UserService for UserServiceImpl {
         }))
     }
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn drop_user(
         &self,
         request: Request<DropUserRequest>,
@@ -142,7 +142,7 @@ impl UserService for UserServiceImpl {
         }))
     }
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn update_user(
         &self,
         request: Request<UpdateUserRequest>,
@@ -165,7 +165,7 @@ impl UserService for UserServiceImpl {
         }))
     }
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn grant_privilege(
         &self,
         request: Request<GrantPrivilegeRequest>,
@@ -185,7 +185,7 @@ impl UserService for UserServiceImpl {
         }))
     }
 
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     async fn revoke_privilege(
         &self,
         request: Request<RevokePrivilegeRequest>,
diff --git a/src/meta/src/hummock/manager/mod.rs b/src/meta/src/hummock/manager/mod.rs
index 2b0c3e3db87dc..1b3a284e9ccc9 100644
--- a/src/meta/src/hummock/manager/mod.rs
+++ b/src/meta/src/hummock/manager/mod.rs
@@ -1761,7 +1761,7 @@ impl HummockManager {
     }
 
     /// Get version deltas from meta store
-    #[cfg_attr(coverage, no_coverage)]
+    #[cfg_attr(coverage, coverage(off))]
     #[named]
     pub async fn list_version_deltas(
         &self,
diff --git a/src/meta/src/lib.rs b/src/meta/src/lib.rs
index afe66d27ad8e8..f549578f079c6 100644
--- a/src/meta/src/lib.rs
+++ b/src/meta/src/lib.rs
@@ -26,13 +26,12 @@
 #![feature(error_generic_member_access)]
 #![feature(assert_matches)]
 #![feature(try_blocks)]
-#![cfg_attr(coverage, feature(no_coverage))]
+#![cfg_attr(coverage, feature(coverage_attribute))]
 #![feature(custom_test_frameworks)]
 #![test_runner(risingwave_test_runner::test_runner::run_failpont_tests)]
 #![feature(is_sorted)]
 #![feature(impl_trait_in_assoc_type)]
 #![feature(type_name_of_val)]
-#![feature(async_fn_in_trait)]
 
 pub mod backup_restore;
 pub mod barrier;
diff --git a/src/prost/helpers/src/lib.rs b/src/prost/helpers/src/lib.rs
index f4d1d1a45baa1..5796e14273fe9 100644
--- a/src/prost/helpers/src/lib.rs
+++ b/src/prost/helpers/src/lib.rs
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![cfg_attr(coverage, feature(no_coverage))]
+#![cfg_attr(coverage, feature(coverage_attribute))]
 #![feature(iterator_try_collect)]
 
 use proc_macro::TokenStream;
@@ -24,7 +24,7 @@ mod generate;
 
 /// This attribute will be placed before any pb types, including messages and enums.
 /// See `prost/helpers/README.md` for more details.
-#[cfg_attr(coverage, no_coverage)]
+#[cfg_attr(coverage, coverage(off))]
 #[proc_macro_derive(AnyPB)]
 pub fn any_pb(input: TokenStream) -> TokenStream {
     // Parse the string representation
@@ -37,7 +37,7 @@ pub fn any_pb(input: TokenStream) -> TokenStream {
 }
 
 // Procedure macros can not be tested from the same crate.
-#[cfg_attr(coverage, no_coverage)]
+#[cfg_attr(coverage, coverage(off))]
 fn produce(ast: &DeriveInput) -> Result<TokenStream2> {
     let name = &ast.ident;
 
diff --git a/src/rpc_client/src/lib.rs b/src/rpc_client/src/lib.rs
index 3e744bb61608d..be871521d3bc4 100644
--- a/src/rpc_client/src/lib.rs
+++ b/src/rpc_client/src/lib.rs
@@ -19,7 +19,7 @@
 #![feature(result_option_inspect)]
 #![feature(type_alias_impl_trait)]
 #![feature(associated_type_defaults)]
-#![feature(generators)]
+#![feature(coroutines)]
 #![feature(iterator_try_collect)]
 #![feature(hash_extract_if)]
 #![feature(try_blocks)]
diff --git a/src/source/src/lib.rs b/src/source/src/lib.rs
index 1a32888cdf651..aaa045c607c95 100644
--- a/src/source/src/lib.rs
+++ b/src/source/src/lib.rs
@@ -16,7 +16,7 @@
 #![feature(trait_alias)]
 #![feature(lint_reasons)]
 #![feature(result_option_inspect)]
-#![feature(generators)]
+#![feature(coroutines)]
 #![feature(hash_extract_if)]
 #![feature(type_alias_impl_trait)]
 #![feature(box_patterns)]
diff --git a/src/storage/Cargo.toml b/src/storage/Cargo.toml
index c6fc5531acd33..fc01eba294564 100644
--- a/src/storage/Cargo.toml
+++ b/src/storage/Cargo.toml
@@ -25,7 +25,7 @@ dyn-clone = "1.0.14"
 either = "1"
 enum-as-inner = "0.6"
 fail = "0.5"
-foyer = { git = "https://github.com/mrcroxx/foyer", rev = "5d0134b" }
+foyer = { git = "https://github.com/MrCroxx/foyer", rev = "2261151" }
 futures = { version = "0.3", default-features = false, features = ["alloc"] }
 futures-async-stream = { workspace = true }
 hex = "0.4"
diff --git a/src/storage/backup/src/lib.rs b/src/storage/backup/src/lib.rs
index 3e0549db188a2..1daacbf691c0d 100644
--- a/src/storage/backup/src/lib.rs
+++ b/src/storage/backup/src/lib.rs
@@ -25,7 +25,7 @@
 #![feature(lazy_cell)]
 #![feature(let_chains)]
 #![feature(error_generic_member_access)]
-#![cfg_attr(coverage, feature(no_coverage))]
+#![cfg_attr(coverage, feature(coverage_attribute))]
 
 pub mod error;
 pub mod meta_snapshot;
diff --git a/src/storage/hummock_test/Cargo.toml b/src/storage/hummock_test/Cargo.toml
index 600a5249ddf1b..8abf2f45e6855 100644
--- a/src/storage/hummock_test/Cargo.toml
+++ b/src/storage/hummock_test/Cargo.toml
@@ -20,7 +20,7 @@ bytes = { version = "1" }
 clap = { version = "4", features = ["derive"] }
 fail = "0.5"
 futures = { version = "0.3", default-features = false, features = ["alloc"] }
-futures-async-stream = "0.2"
+futures-async-stream = "0.2.9"
 itertools = "0.11"
 parking_lot = "0.12"
 rand = "0.8"
@@ -47,7 +47,7 @@ futures = { version = "0.3", default-features = false, features = [
     "executor",
 ] }
 
-futures-async-stream = "0.2"
+futures-async-stream = "0.2.9"
 risingwave_test_runner = { workspace = true }
 serial_test = "2.0"
 sync-point = { path = "../../utils/sync-point" }
diff --git a/src/storage/hummock_test/src/bin/replay/main.rs b/src/storage/hummock_test/src/bin/replay/main.rs
index 1e9c9591bc864..ae6038d8b5d16 100644
--- a/src/storage/hummock_test/src/bin/replay/main.rs
+++ b/src/storage/hummock_test/src/bin/replay/main.rs
@@ -13,7 +13,7 @@
 // limitations under the License.
 
 #![feature(bound_map)]
-#![feature(generators)]
+#![feature(coroutines)]
 #![feature(stmt_expr_attributes)]
 #![feature(proc_macro_hygiene)]
 
diff --git a/src/storage/hummock_test/src/lib.rs b/src/storage/hummock_test/src/lib.rs
index 73e1d8cd0eaad..593771435f1e0 100644
--- a/src/storage/hummock_test/src/lib.rs
+++ b/src/storage/hummock_test/src/lib.rs
@@ -17,7 +17,6 @@
 #![feature(bound_map)]
 #![feature(type_alias_impl_trait)]
 #![feature(associated_type_bounds)]
-#![feature(return_position_impl_trait_in_trait)]
 
 #[cfg(test)]
 mod compactor_tests;
diff --git a/src/storage/hummock_trace/Cargo.toml b/src/storage/hummock_trace/Cargo.toml
index 46eabf17835e4..150b35b79cda0 100644
--- a/src/storage/hummock_trace/Cargo.toml
+++ b/src/storage/hummock_trace/Cargo.toml
@@ -14,7 +14,7 @@ bincode = { version = "=2.0.0-rc.3", features = ["serde"] }
 byteorder = "1"
 bytes = { version = "1", features = ["serde"] }
 futures = { version = "0.3", default-features = false, features = ["alloc"] }
-futures-async-stream = "0.2"
+futures-async-stream = "0.2.9"
 parking_lot = "0.12"
 prost = { workspace = true }
 risingwave_common = { workspace = true }
diff --git a/src/storage/hummock_trace/src/lib.rs b/src/storage/hummock_trace/src/lib.rs
index df757c58cc4fa..8c6c8913205ab 100644
--- a/src/storage/hummock_trace/src/lib.rs
+++ b/src/storage/hummock_trace/src/lib.rs
@@ -16,7 +16,7 @@
 #![feature(cursor_remaining)]
 #![feature(bound_map)]
 #![feature(trait_alias)]
-#![feature(generators)]
+#![feature(coroutines)]
 
 mod collector;
 mod error;
diff --git a/src/storage/src/hummock/compactor/compactor_runner.rs b/src/storage/src/hummock/compactor/compactor_runner.rs
index 583bab3d10b3c..a21016014d247 100644
--- a/src/storage/src/hummock/compactor/compactor_runner.rs
+++ b/src/storage/src/hummock/compactor/compactor_runner.rs
@@ -140,9 +140,6 @@ impl CompactorRunner {
         Ok((self.split_index, ssts, compaction_stat))
     }
 
-    // This is a clippy bug, see https://github.com/rust-lang/rust-clippy/issues/11380.
-    // TODO: remove `allow` here after the issued is closed.
-    #[expect(clippy::needless_pass_by_ref_mut)]
     pub async fn build_delete_range_iter<F: CompactionFilter>(
         sstable_infos: &Vec<SstableInfo>,
         sstable_store: &SstableStoreRef,
diff --git a/src/storage/src/hummock/compactor/mod.rs b/src/storage/src/hummock/compactor/mod.rs
index d2f36167675e7..137682d6f7825 100644
--- a/src/storage/src/hummock/compactor/mod.rs
+++ b/src/storage/src/hummock/compactor/mod.rs
@@ -325,7 +325,7 @@ impl Compactor {
 
 /// The background compaction thread that receives compaction tasks from hummock compaction
 /// manager and runs compaction tasks.
-#[cfg_attr(coverage, no_coverage)]
+#[cfg_attr(coverage, coverage(off))]
 pub fn start_compactor(
     compactor_context: CompactorContext,
     hummock_meta_client: Arc<dyn HummockMetaClient>,
@@ -618,7 +618,7 @@ pub fn start_compactor(
 
 /// The background compaction thread that receives compaction tasks from hummock compaction
 /// manager and runs compaction tasks.
-#[cfg_attr(coverage, no_coverage)]
+#[cfg_attr(coverage, coverage(off))]
 pub fn start_shared_compactor(
     grpc_proxy_client: GrpcCompactorProxyClient,
     mut receiver: mpsc::UnboundedReceiver<Request<DispatchCompactionTaskRequest>>,
diff --git a/src/storage/src/hummock/event_handler/uploader.rs b/src/storage/src/hummock/event_handler/uploader.rs
index 995a9d181e2f5..a07da55fb7046 100644
--- a/src/storage/src/hummock/event_handler/uploader.rs
+++ b/src/storage/src/hummock/event_handler/uploader.rs
@@ -1641,9 +1641,6 @@ mod tests {
         (buffer_tracker, uploader, new_task_notifier)
     }
 
-    // This is a clippy bug, see https://github.com/rust-lang/rust-clippy/issues/11380.
-    // TODO: remove `allow` here after the issued is closed.
-    #[expect(clippy::needless_pass_by_ref_mut)]
     async fn assert_uploader_pending(uploader: &mut HummockUploader) {
         for _ in 0..10 {
             yield_now().await;
diff --git a/src/storage/src/lib.rs b/src/storage/src/lib.rs
index c5ffe656ab893..0f2f155f6a903 100644
--- a/src/storage/src/lib.rs
+++ b/src/storage/src/lib.rs
@@ -18,7 +18,7 @@
 #![feature(bound_map)]
 #![feature(custom_test_frameworks)]
 #![feature(extract_if)]
-#![feature(generators)]
+#![feature(coroutines)]
 #![feature(hash_extract_if)]
 #![feature(lint_reasons)]
 #![feature(proc_macro_hygiene)]
@@ -35,15 +35,13 @@
 #![feature(btree_extract_if)]
 #![feature(exact_size_is_empty)]
 #![feature(lazy_cell)]
-#![cfg_attr(coverage, feature(no_coverage))]
+#![cfg_attr(coverage, feature(coverage_attribute))]
 #![recursion_limit = "256"]
 #![feature(error_generic_member_access)]
 #![feature(let_chains)]
 #![feature(associated_type_bounds)]
 #![feature(exclusive_range_pattern)]
 #![feature(impl_trait_in_assoc_type)]
-#![feature(async_fn_in_trait)]
-#![feature(return_position_impl_trait_in_trait)]
 
 pub mod hummock;
 pub mod memory;
diff --git a/src/stream/src/common/table/state_table_cache.rs b/src/stream/src/common/table/state_table_cache.rs
index 156637a41a1a4..b458ef52537e4 100644
--- a/src/stream/src/common/table/state_table_cache.rs
+++ b/src/stream/src/common/table/state_table_cache.rs
@@ -67,9 +67,9 @@ type WatermarkCacheKey = DefaultOrdered<OwnedRow>;
 ///        Issue delete ranges.
 ///
 ///    B. Refreshing the cache:
-///        On barrier, do table scan from most_recently_cleaned_watermark (inclusive) to +inf.
+///        On barrier, do table scan from `most_recently_cleaned_watermark` (inclusive) to +inf.
 ///        Take the Top N rows and insert into cache.
-///        This has to be implemented in state_table.
+///        This has to be implemented in `state_table`.
 ///        We do not need to store any values, just the keys.
 ///
 /// TODO(kwannoel):
diff --git a/src/stream/src/executor/backfill/utils.rs b/src/stream/src/executor/backfill/utils.rs
index 259b67d5f202b..663f9be94cf5e 100644
--- a/src/stream/src/executor/backfill/utils.rs
+++ b/src/stream/src/executor/backfill/utils.rs
@@ -309,9 +309,6 @@ pub(crate) async fn get_progress_per_vnode<S: StateStore, const IS_REPLICATED: b
 }
 
 /// Flush the data
-// This is a clippy bug, see https://github.com/rust-lang/rust-clippy/issues/11380.
-// TODO: remove `allow` here after the issued is closed.
-#[expect(clippy::needless_pass_by_ref_mut)]
 pub(crate) async fn flush_data<S: StateStore, const IS_REPLICATED: bool>(
     table: &mut StateTableInner<S, BasicSerde, IS_REPLICATED>,
     epoch: EpochPair,
diff --git a/src/stream/src/executor/hash_join.rs b/src/stream/src/executor/hash_join.rs
index 4178012cb9d9e..75414fe24a379 100644
--- a/src/stream/src/executor/hash_join.rs
+++ b/src/stream/src/executor/hash_join.rs
@@ -202,11 +202,11 @@ impl<K: HashKey, S: StateStore> std::fmt::Debug for JoinSide<K, S> {
 
 impl<K: HashKey, S: StateStore> JoinSide<K, S> {
     // WARNING: Please do not call this until we implement it.
-    #[expect(dead_code)]
     fn is_dirty(&self) -> bool {
         unimplemented!()
     }
 
+    #[expect(dead_code)]
     fn clear_cache(&mut self) {
         assert!(
             !self.is_dirty(),
diff --git a/src/stream/src/lib.rs b/src/stream/src/lib.rs
index 5a68b1b712b26..389dfae7b8c0c 100644
--- a/src/stream/src/lib.rs
+++ b/src/stream/src/lib.rs
@@ -21,8 +21,8 @@
 #![feature(let_chains)]
 #![feature(hash_extract_if)]
 #![feature(extract_if)]
-#![feature(generators)]
-#![feature(iter_from_generator)]
+#![feature(coroutines)]
+#![feature(iter_from_coroutine)]
 #![feature(proc_macro_hygiene)]
 #![feature(stmt_expr_attributes)]
 #![feature(allocator_api)]
@@ -36,13 +36,11 @@
 #![feature(bound_map)]
 #![feature(iter_order_by)]
 #![feature(exact_size_is_empty)]
-#![feature(return_position_impl_trait_in_trait)]
 #![feature(impl_trait_in_assoc_type)]
 #![feature(test)]
 #![feature(is_sorted)]
 #![feature(btree_cursors)]
 #![feature(assert_matches)]
-#![feature(async_fn_in_trait)]
 
 #[macro_use]
 extern crate tracing;
diff --git a/src/stream/tests/integration_tests/hash_agg.rs b/src/stream/tests/integration_tests/hash_agg.rs
index 1b61bc5cd1d7f..9f4908f252532 100644
--- a/src/stream/tests/integration_tests/hash_agg.rs
+++ b/src/stream/tests/integration_tests/hash_agg.rs
@@ -284,7 +284,7 @@ async fn test_hash_agg_emit_on_window_close() {
     };
 
     check_with_script(
-        || create_executor(),
+        create_executor,
         &format!(
             r###"
             - !barrier 1
diff --git a/src/tests/compaction_test/src/bin/compaction.rs b/src/tests/compaction_test/src/bin/compaction.rs
index 443b79ad625b8..d9ba16f7437b8 100644
--- a/src/tests/compaction_test/src/bin/compaction.rs
+++ b/src/tests/compaction_test/src/bin/compaction.rs
@@ -12,9 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![cfg_attr(coverage, feature(no_coverage))]
+#![cfg_attr(coverage, feature(coverage_attribute))]
 
-#[cfg_attr(coverage, no_coverage)]
+#[cfg_attr(coverage, coverage(off))]
 fn main() {
     use clap::Parser;
 
diff --git a/src/tests/compaction_test/src/bin/delete_range.rs b/src/tests/compaction_test/src/bin/delete_range.rs
index 348a71dc3cce5..592f61a3db4fa 100644
--- a/src/tests/compaction_test/src/bin/delete_range.rs
+++ b/src/tests/compaction_test/src/bin/delete_range.rs
@@ -12,9 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![cfg_attr(coverage, feature(no_coverage))]
+#![cfg_attr(coverage, feature(coverage_attribute))]
 
-#[cfg_attr(coverage, no_coverage)]
+#[cfg_attr(coverage, coverage(off))]
 fn main() {
     use clap::Parser;
 
diff --git a/src/utils/pgwire/src/lib.rs b/src/utils/pgwire/src/lib.rs
index 1cda373ee9568..84a17d9907879 100644
--- a/src/utils/pgwire/src/lib.rs
+++ b/src/utils/pgwire/src/lib.rs
@@ -17,8 +17,6 @@
 #![feature(result_option_inspect)]
 #![feature(iterator_try_collect)]
 #![feature(trusted_len)]
-#![feature(async_fn_in_trait)]
-#![feature(return_position_impl_trait_in_trait)]
 #![feature(lazy_cell)]
 #![expect(clippy::doc_markdown, reason = "FIXME: later")]
 
diff --git a/src/workspace-hack/Cargo.toml b/src/workspace-hack/Cargo.toml
index 6c08e08490f7d..67b218c787652 100644
--- a/src/workspace-hack/Cargo.toml
+++ b/src/workspace-hack/Cargo.toml
@@ -37,7 +37,7 @@ combine = { version = "4", features = ["tokio"] }
 crossbeam-epoch = { version = "0.9" }
 crossbeam-queue = { version = "0.3" }
 crossbeam-utils = { version = "0.8" }
-deranged = { version = "0.3", default-features = false, features = ["serde", "std"] }
+deranged = { version = "0.3", default-features = false, features = ["powerfmt", "serde", "std"] }
 digest = { version = "0.10", features = ["mac", "oid", "std"] }
 either = { version = "1", features = ["serde"] }
 fail = { version = "0.5", default-features = false, features = ["failpoints"] }
@@ -147,7 +147,7 @@ auto_enums = { version = "0.8", features = ["futures03"] }
 bitflags = { version = "2", default-features = false, features = ["serde", "std"] }
 bytes = { version = "1", features = ["serde"] }
 cc = { version = "1", default-features = false, features = ["parallel"] }
-deranged = { version = "0.3", default-features = false, features = ["serde", "std"] }
+deranged = { version = "0.3", default-features = false, features = ["powerfmt", "serde", "std"] }
 either = { version = "1", features = ["serde"] }
 fixedbitset = { version = "0.4" }
 frunk_core = { version = "0.4", default-features = false, features = ["std"] }