address warnings and such

pageserver/client/src/page_service.rs

@@ -118,11 +118,13 @@ pub struct PagestreamClient {
 }
 
 pub struct PagestreamSender {
+    #[allow(dead_code)]
     shared: Arc<Mutex<PagestreamShared>>,
     sink: SplitSink<tokio_postgres::CopyBothDuplex<bytes::Bytes>, bytes::Bytes>,
 }
 
 pub struct PagestreamReceiver {
+    #[allow(dead_code)]
     shared: Arc<Mutex<PagestreamShared>>,
     stream: SplitStream<tokio_postgres::CopyBothDuplex<bytes::Bytes>>,
 }
@@ -188,7 +190,7 @@ impl PagestreamClient {
 
     pub fn split(self) -> (PagestreamSender, PagestreamReceiver) {
         let Self {
-            shared,
+            shared: _,
             sink,
             stream,
         } = self;

pageserver/src/bin/test_helper_slow_client_reads.rs

@@ -4,11 +4,10 @@ use std::{
 };
 
 use clap::Parser;
-use pageserver_api::models::{PagestreamGetPageRequest, PagestreamRequest, PagestreamTestRequest};
+use pageserver_api::models::{PagestreamRequest, PagestreamTestRequest};
 use utils::{
     id::{TenantId, TimelineId},
     lsn::Lsn,
-    shard::TenantShardId,
 };
 
 #[derive(clap::Parser)]
@@ -27,7 +26,7 @@ async fn main() -> anyhow::Result<()> {
     } = Args::parse();
     let client = pageserver_client::page_service::Client::new(connstr).await?;
     let client = client.pagestream(tenant_id, timeline_id).await?;
-    let (mut sender, mut receiver) = client.split();
+    let (mut sender, _receiver) = client.split();
 
     eprintln!("filling the pipe");
     let mut msg = 0;
@@ -44,7 +43,7 @@ async fn main() -> anyhow::Result<()> {
                 message: format!("message {}", msg),
             },
         ));
-        let Ok(res) = tokio::time::timeout(Duration::from_secs(1), fut).await else {
+        let Ok(res) = tokio::time::timeout(Duration::from_secs(10), fut).await else {
             eprintln!("pipe seems full");
             break;
         };
@@ -54,6 +53,8 @@ async fn main() -> anyhow::Result<()> {
     stdout().write(b"R")?;
     stdout().flush()?;
 
+    eprintln!("waiting for signal to tell us to exit");
+
     let mut buf = [0u8; 1];
     stdin().read_exact(&mut buf)?;
 

pageserver/src/page_service.rs

@@ -39,7 +39,6 @@ use tokio::io::{AsyncWriteExt, BufWriter};
 use tokio::task::JoinHandle;
 use tokio_util::sync::CancellationToken;
 use tracing::*;
-use utils::failpoint_support;
 use utils::sync::spsc_fold;
 use utils::{
     auth::{Claims, Scope, SwappableJwtAuth},
@@ -1221,10 +1220,7 @@ impl PageServerHandler {
             });
 
             // what we want to do
-            let flush_fut = async {
-                failpoint_support::sleep_millis_async!("page_service:flush:pre");
-                pgb_writer.flush().await
-            };
+            let flush_fut = pgb_writer.flush();
             // metric for how long flushing takes
             let flush_fut = match flushing_timer {
                 Some(flushing_timer) => {
@@ -1873,28 +1869,25 @@ impl PageServerHandler {
         ))
     }
 
+    // NB: this impl mimics what we do for batched getpage requests.
     #[cfg(feature = "testing")]
     #[instrument(skip_all, fields(shard_id))]
     async fn handle_test_request_batch(
         &mut self,
         timeline: &Timeline,
         requests: Vec<BatchedTestRequest>,
-        ctx: &RequestContext,
+        _ctx: &RequestContext,
     ) -> Vec<Result<(PagestreamBeMessage, SmgrOpTimer), BatchedPageStreamError>> {
-        use pageserver_api::models::PagestreamTestResponse;
-
-        let latest_gc_cutoff_lsn = timeline.get_latest_gc_cutoff_lsn();
-
         // real requests would do something with the timeline
         let mut results = Vec::with_capacity(requests.len());
-        for req in requests.iter() {
+        for _req in requests.iter() {
             tokio::task::yield_now().await;
 
             results.push({
                 if timeline.cancel.is_cancelled() {
                     Err(PageStreamError::Shutdown)
                 } else {
-                    Ok(PagestreamTestResponse { req: req.req.clone() })
+                    Ok(())
                 }
             });
         }
@@ -1905,7 +1898,7 @@ impl PageServerHandler {
                 .into_iter()
                 .zip(results.into_iter())
                 .map(|(req, res)| {
-                    res.map(|page| {
+                    res.map(|()| {
                         (
                             PagestreamBeMessage::Test(models::PagestreamTestResponse {
                                 req: req.req.clone(),

test_runner/regress/test_page_service_batching_regressions.py

@@ -19,8 +19,6 @@ def test_slow_flush(neon_env_builder: NeonEnvBuilder, neon_binpath: Path):
     env = neon_env_builder.init_start()
 
     log.info("make flush appear slow")
-    ps_http = env.pageserver.http_client()
-    ps_http.configure_failpoints(("page_service:flush:pre", "return(10000000)"))
 
     log.info("filling pipe")
     child = subprocess.Popen(