Revert "undo that last piece"

This reverts commit e3481bfcae.

pageserver/src/metrics.rs

@@ -1223,7 +1223,8 @@ pub(crate) mod virtual_file_io_engine {
     });
 }
 
-pub(crate) struct SmgrOpTimer {
+pub(crate) struct SmgrOpTimer(Option<SmgrOpTimerInner>);
+pub(crate) struct SmgrOpTimerInner {
     global_latency_histo: Histogram,
 
     // Optional because not all op types are tracked per-timeline
@@ -1239,13 +1240,19 @@ impl SmgrOpTimer {
         let Some(throttle) = throttle else {
             return;
         };
-        self.throttled += *throttle;
+        let inner = self.0.as_mut().expect("other public methods consume self");
+        inner.throttled += *throttle;
     }
-}
 
-impl Drop for SmgrOpTimer {
-    fn drop(&mut self) {
-        let inner = self;
+    pub(crate) fn observe_smgr_op_completion_and_start_flushing(mut self) {
+        let (flush_start, inner) = self
+            .smgr_op_end()
+            .expect("this method consume self, and the only other caller is drop handler");
+    }
+
+    /// Returns `None`` if this method has already been called, `Some` otherwise.
+    fn smgr_op_end(&mut self) -> Option<(Instant, SmgrOpTimerInner)> {
+        let inner = self.0.take()?;
 
         let now = Instant::now();
         let elapsed = now - inner.start;
@@ -1275,6 +1282,14 @@ impl Drop for SmgrOpTimer {
         if let Some(per_timeline_getpage_histo) = &inner.per_timeline_latency_histo {
             per_timeline_getpage_histo.observe(elapsed);
         }
+
+        Some((now, inner))
+    }
+}
+
+impl Drop for SmgrOpTimer {
+    fn drop(&mut self) {
+        self.smgr_op_end();
     }
 }
 
@@ -1556,13 +1571,13 @@ impl SmgrQueryTimePerTimeline {
             None
         };
 
-        SmgrOpTimer {
+        SmgrOpTimer(Some(SmgrOpTimerInner {
             global_latency_histo: self.global_latency[op as usize].clone(),
             per_timeline_latency_histo,
             start: started_at,
             op,
             throttled: Duration::ZERO,
-        }
+        }))
     }
 
     pub(crate) fn observe_getpage_batch_start(&self, batch_size: usize) {

pageserver/src/page_service.rs

@@ -1077,6 +1077,8 @@ impl PageServerHandler {
             // The timer's underlying metric is used for a storage-internal latency SLO and
             // we don't want to include latency in it that we can't control.
             // And as pointed out above, in this case, we don't control the time that flush will take.
+            let flushing_timer =
+                timer.map(|timer| timer.observe_smgr_op_completion_and_start_flushing());
 
             // what we want to do
             let flush_fut = pgb_writer.flush();
@@ -1098,8 +1100,6 @@ impl PageServerHandler {
             // and log the info! line inside the request span
             // .instrument(span.clone())
             .await?;
-
-            drop(timer);
         }
         Ok(())
     }