From af95320a8c31ee0713199448e5ab9f9c0728e37c Mon Sep 17 00:00:00 2001
From: Christian Schwarz <christian@neon.tech>
Date: Wed, 20 Nov 2024 14:25:05 +0100
Subject: [PATCH] Revert "Revert "switch back to tokio::time::sleep, to get the
 numbers""

This reverts commit aa695b2ad7d0272164255b8d992c6a74f66e45e3.
---
 Cargo.lock                     | 12 ------------
 Cargo.toml                     |  1 -
 pageserver/Cargo.toml          |  1 -
 pageserver/src/page_service.rs | 17 +++--------------
 4 files changed, 3 insertions(+), 28 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 9196015057e4..c7af140f7df2 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -244,17 +244,6 @@ dependencies = [
  "syn 2.0.52",
 ]
 
-[[package]]
-name = "async-timer"
-version = "0.7.4"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ba5fa6ed76cb2aa820707b4eb9ec46f42da9ce70b0eafab5e5e34942b38a44d5"
-dependencies = [
- "libc",
- "wasm-bindgen",
- "winapi",
-]
-
 [[package]]
 name = "async-trait"
 version = "0.1.68"
@@ -3601,7 +3590,6 @@ dependencies = [
  "arc-swap",
  "async-compression",
  "async-stream",
- "async-timer",
  "bit_field",
  "byteorder",
  "bytes",
diff --git a/Cargo.toml b/Cargo.toml
index d74efd51f83b..dbda9305351f 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -47,7 +47,6 @@ anyhow = { version = "1.0", features = ["backtrace"] }
 arc-swap = "1.6"
 async-compression = { version = "0.4.0", features = ["tokio", "gzip", "zstd"] }
 atomic-take = "1.1.0"
-async-timer = "0.7.4"
 azure_core = { version = "0.19", default-features = false, features = ["enable_reqwest_rustls", "hmac_rust"] }
 azure_identity = { version = "0.19", default-features = false, features = ["enable_reqwest_rustls"] }
 azure_storage = { version = "0.19", default-features = false, features = ["enable_reqwest_rustls"] }
diff --git a/pageserver/Cargo.toml b/pageserver/Cargo.toml
index ed2ee2f5c29f..143d8236dff5 100644
--- a/pageserver/Cargo.toml
+++ b/pageserver/Cargo.toml
@@ -15,7 +15,6 @@ anyhow.workspace = true
 arc-swap.workspace = true
 async-compression.workspace = true
 async-stream.workspace = true
-async-timer.workspace = true
 bit_field.workspace = true
 byteorder.workspace = true
 bytes.workspace = true
diff --git a/pageserver/src/page_service.rs b/pageserver/src/page_service.rs
index c32f889958c1..3472dda378d6 100644
--- a/pageserver/src/page_service.rs
+++ b/pageserver/src/page_service.rs
@@ -3,7 +3,6 @@
 
 use anyhow::{bail, Context};
 use async_compression::tokio::write::GzipEncoder;
-use async_timer::Oneshot;
 use bytes::Buf;
 use futures::FutureExt;
 use itertools::Itertools;
@@ -23,7 +22,6 @@ use pq_proto::FeStartupPacket;
 use pq_proto::{BeMessage, FeMessage, RowDescriptor};
 use std::borrow::Cow;
 use std::io;
-use std::pin::Pin;
 use std::str;
 use std::str::FromStr;
 use std::sync::Arc;
@@ -318,8 +316,6 @@ struct PageServerHandler {
 
     /// See [`PageServerConf::server_side_batch_timeout`]
     server_side_batch_timeout: Option<Duration>,
-
-    server_side_batch_timer: Pin<Box<async_timer::oneshot::Timer>>,
 }
 
 struct Carry {
@@ -589,9 +585,6 @@ impl PageServerHandler {
             timeline_handles: TimelineHandles::new(tenant_manager),
             cancel,
             server_side_batch_timeout,
-            server_side_batch_timer: Box::pin(async_timer::oneshot::Timer::new(
-                Duration::from_secs(999),
-            )), // reset each iteration
         }
     }
 
@@ -668,13 +661,9 @@ impl PageServerHandler {
                                         .msg
                                 ]));
                             } else {
-                                std::future::poll_fn(|ctx| {
-                                    self.server_side_batch_timer
-                                        .restart(batch_timeout, ctx.waker());
-                                    std::task::Poll::Ready(())
-                                })
-                                .await;
-                                batching_deadline_storage = Some(&mut self.server_side_batch_timer);
+                                batching_deadline_storage = Some(Box::pin(async move {
+                                    tokio::time::sleep(batch_timeout).await;
+                                }));
                                 Either::Right(
                                     batching_deadline_storage.as_mut().expect("we just set it"),
                                 )