Revert "[persist] Always perform routine maintenance (#16654)"

This reverts commit 0e1aaa0.

Author: guswynn

src/persist-client/src/cli/admin.rs

@@ -248,12 +248,9 @@ pub async fn force_compaction(
                     .sum::<usize>(),
                 start.elapsed(),
             );
-            let (apply_res, maintenance) = machine
+            let apply_res = machine
                 .merge_res(&FueledMergeRes { output: res.output })
                 .await;
-            if !maintenance.is_empty() {
-                info!("ignoring non-empty requested maintenance: {maintenance:?}")
-            }
             match apply_res {
                 ApplyMergeResult::AppliedExact | ApplyMergeResult::AppliedSubset => {
                     info!("attempt {} req {}: {:?}", attempt, idx, apply_res);

src/persist-client/src/critical.rs

@@ -297,8 +297,7 @@ where
     /// Politely expires this reader, releasing its since capability.
     #[instrument(level = "debug", skip_all, fields(shard = %self.machine.shard_id()))]
     pub async fn expire(mut self) {
-        let (_, maintenance) = self.machine.expire_critical_reader(&self.reader_id).await;
-        maintenance.start_performing(&self.machine, &self.gc);
+        self.machine.expire_critical_reader(&self.reader_id).await;
     }
 }
 

src/persist-client/src/internal/compact.rs

@@ -40,7 +40,6 @@ use tracing::{debug, debug_span, trace, Instrument, Span};
 use crate::async_runtime::CpuHeavyRuntime;
 use crate::batch::BatchParts;
 use crate::fetch::{fetch_batch_part, EncodedPart};
-use crate::internal::gc::GarbageCollector;
 use crate::internal::machine::{retry_external, Machine};
 use crate::internal::state::{HollowBatch, HollowBatchPart};
 use crate::internal::trace::{ApplyMergeResult, FueledMergeRes};
@@ -74,7 +73,7 @@ pub struct CompactRes<T> {
 /// This will possibly be called over RPC in the future. Physical compaction is
 /// merging adjacent batches. Logical compaction is advancing timestamps to a
 /// new since and consolidating the resulting updates.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct Compactor<K, V, T, D> {
     cfg: PersistConfig,
     metrics: Arc<Metrics>,
@@ -87,30 +86,18 @@ pub struct Compactor<K, V, T, D> {
     _phantom: PhantomData<fn() -> D>,
 }
 
-impl<K, V, T, D> Clone for Compactor<K, V, T, D> {
-    fn clone(&self) -> Self {
-        Compactor {
-            cfg: self.cfg.clone(),
-            metrics: Arc::clone(&self.metrics),
-            sender: self.sender.clone(),
-            _phantom: Default::default(),
-        }
-    }
-}
-
 impl<K, V, T, D> Compactor<K, V, T, D>
 where
     K: Debug + Codec,
     V: Debug + Codec,
     T: Timestamp + Lattice + Codec64,
-    D: Semigroup + Codec64 + Send + Sync,
+    D: Semigroup + Codec64 + Send,
 {
     pub fn new(
         cfg: PersistConfig,
         metrics: Arc<Metrics>,
         cpu_heavy_runtime: Arc<CpuHeavyRuntime>,
         writer_id: WriterId,
-        gc: GarbageCollector<K, V, T, D>,
     ) -> Self {
         let (compact_req_sender, mut compact_req_receiver) = mpsc::channel::<(
             Instant,
@@ -165,7 +152,6 @@ where
                 let compact_span =
                     debug_span!(parent: None, "compact::apply", shard_id=%machine.shard_id());
                 compact_span.follows_from(&Span::current());
-                let gc = gc.clone();
                 let _ = mz_ore::task::spawn(|| "PersistCompactionWorker", async move {
                     let res = Self::compact_and_apply(
                         cfg,
@@ -175,7 +161,6 @@ where
                         req,
                         writer_id,
                         &mut machine,
-                        &gc,
                     )
                     .instrument(compact_span)
                     .await;
@@ -253,7 +238,6 @@ where
         req: CompactReq<T>,
         writer_id: WriterId,
         machine: &mut Machine<K, V, T, D>,
-        gc: &GarbageCollector<K, V, T, D>,
     ) -> Result<ApplyMergeResult, anyhow::Error> {
         metrics.compaction.started.inc();
         let start = Instant::now();
@@ -317,8 +301,7 @@ where
         match res {
             Ok(Ok(res)) => {
                 let res = FueledMergeRes { output: res.output };
-                let (apply_merge_result, maintenance) = machine.merge_res(&res).await;
-                maintenance.start_performing(machine, gc);
+                let apply_merge_result = machine.merge_res(&res).await;
                 match &apply_merge_result {
                     ApplyMergeResult::AppliedExact => {
                         metrics.compaction.applied.inc();

src/persist-client/src/internal/gc.rs

@@ -10,7 +10,6 @@
 use std::collections::HashSet;
 use std::fmt::Debug;
 use std::marker::PhantomData;
-use std::mem;
 use std::time::Instant;
 
 use differential_dataflow::difference::Semigroup;
@@ -24,19 +23,19 @@ use tokio::sync::{mpsc, oneshot};
 use tracing::{debug, debug_span, warn, Instrument, Span};
 
 use crate::internal::machine::{retry_external, Machine};
-use crate::internal::maintenance::RoutineMaintenance;
 use crate::internal::paths::{PartialRollupKey, RollupId};
 use crate::ShardId;
 
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, Clone)]
+#[cfg_attr(test, derive(PartialEq))]
 pub struct GcReq {
     pub shard_id: ShardId,
     pub new_seqno_since: SeqNo,
 }
 
 #[derive(Debug)]
 pub struct GarbageCollector<K, V, T, D> {
-    sender: UnboundedSender<(GcReq, oneshot::Sender<RoutineMaintenance>)>,
+    sender: UnboundedSender<(GcReq, oneshot::Sender<()>)>,
     _phantom: PhantomData<fn() -> (K, V, T, D)>,
 }
 
@@ -106,7 +105,7 @@ where
 {
     pub fn new(mut machine: Machine<K, V, T, D>) -> Self {
         let (gc_req_sender, mut gc_req_recv) =
-            mpsc::unbounded_channel::<(GcReq, oneshot::Sender<RoutineMaintenance>)>();
+            mpsc::unbounded_channel::<(GcReq, oneshot::Sender<()>)>();
 
         // spin off a single task responsible for executing GC requests.
         // work is enqueued into the task through a channel
@@ -142,7 +141,7 @@ where
 
                 let start = Instant::now();
                 machine.metrics.gc.started.inc();
-                let mut maintenance = Self::gc_and_truncate(&mut machine, consolidated_req)
+                Self::gc_and_truncate(&mut machine, consolidated_req)
                     .instrument(gc_span)
                     .await;
                 machine.metrics.gc.finished.inc();
@@ -156,9 +155,8 @@ where
                 // inform all callers who enqueued GC reqs that their work is complete
                 for sender in gc_completed_senders {
                     // we can safely ignore errors here, it's possible the caller
-                    // wasn't interested in waiting and dropped their receiver.
-                    // maintenance will be somewhat-arbitrarily assigned to the first oneshot.
-                    let _ = sender.send(mem::take(&mut maintenance));
+                    // wasn't interested in waiting and dropped their receiver
+                    let _ = sender.send(());
                 }
             }
         });
@@ -172,10 +170,7 @@ where
     /// Enqueues a [GcReq] to be consumed by the GC background task when available.
     ///
     /// Returns a future that indicates when GC has cleaned up to at least [GcReq::new_seqno_since]
-    pub fn gc_and_truncate_background(
-        &self,
-        req: GcReq,
-    ) -> Option<oneshot::Receiver<RoutineMaintenance>> {
+    pub fn gc_and_truncate_background(&self, req: GcReq) -> Option<oneshot::Receiver<()>> {
         let (gc_completed_sender, gc_completed_receiver) = oneshot::channel();
         let new_gc_sender = self.sender.clone();
         let send = new_gc_sender.send((req, gc_completed_sender));
@@ -193,10 +188,7 @@ where
         Some(gc_completed_receiver)
     }
 
-    pub async fn gc_and_truncate(
-        machine: &mut Machine<K, V, T, D>,
-        req: GcReq,
-    ) -> RoutineMaintenance {
+    pub async fn gc_and_truncate(machine: &mut Machine<K, V, T, D>, req: GcReq) {
         assert_eq!(req.shard_id, machine.shard_id());
         // NB: Because these requests can be processed concurrently (and in
         // arbitrary order), all of the logic below has to work even if we've
@@ -232,7 +224,7 @@ where
                 "gc {} early returning, already GC'd past {}",
                 req.shard_id, req.new_seqno_since,
             );
-            return RoutineMaintenance::default();
+            return;
         }
 
         let mut deleteable_batch_blobs = HashSet::new();
@@ -312,7 +304,7 @@ where
             .state_versions
             .write_rollup_blob(&machine.shard_metrics, &state, &rollup_key)
             .await;
-        let (applied, maintenance) = machine
+        let applied = machine
             .add_and_remove_rollups((rollup_seqno, &rollup_key), &deleteable_rollup_blobs)
             .await;
         // We raced with some other GC process to write this rollup out. Ours
@@ -359,7 +351,5 @@ where
             "gc {} truncated diffs through seqno {}",
             req.shard_id, req.new_seqno_since
         );
-
-        maintenance
     }
 }