Merge pull request #30114 from teskje/compute-sequence-readhold-downgrades

controller/compute: sequence external read hold changes

Author: teskje

Cargo.lock

@@ -119,7 +119,7 @@ impl TimestampProvider for Frontiers {
 
         let mock_read_hold = |id, frontier| {
             let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
-            ReadHold::new(id, frontier, tx)
+            ReadHold::with_channel(id, frontier, tx)
         };
 
         for (instance_id, ids) in id_bundle.compute_ids.iter() {

src/adapter/tests/timestamp_selection.rs

@@ -17,6 +17,7 @@ bytes = "1.3.0"
 bytesize = "1.1.0"
 chrono = { version = "0.4.35", default-features = false, features = ["std"] }
 crossbeam-channel = "0.5.8"
+derivative = "2.2.0"
 differential-dataflow = "0.13.2"
 futures = "0.3.25"
 http = "1.1.0"

src/compute-client/Cargo.toml

@@ -944,7 +944,7 @@ mod tests {
                     .get(&id)
                     .ok_or(ReadHoldError::CollectionMissing(id))?;
                 let (tx, _rx) = tokio::sync::mpsc::unbounded_channel();
-                holds.push(ReadHold::new(id, read.clone(), tx));
+                holds.push(ReadHold::with_channel(id, read.clone(), tx));
             }
             Ok(holds)
         }

src/compute-client/src/as_of_selection.rs

@@ -58,7 +58,7 @@ use mz_storage_types::read_policy::ReadPolicy;
 use mz_storage_types::time_dependence::{TimeDependence, TimeDependenceError};
 use prometheus::proto::LabelPair;
 use serde::{Deserialize, Serialize};
-use timely::progress::{Antichain, ChangeBatch, Timestamp};
+use timely::progress::{Antichain, Timestamp};
 use timely::PartialOrder;
 use tokio::sync::{mpsc, oneshot};
 use tokio::time::{self, MissedTickBehavior};
@@ -539,8 +539,6 @@ where
             logs.push((log, id, shared));
         }
 
-        let (read_holds_tx, read_holds_rx) = mpsc::unbounded_channel();
-
         let client = instance::Client::spawn(
             id,
             self.build_info,
@@ -553,11 +551,9 @@ where
             Arc::clone(&self.dyncfg),
             self.response_tx.clone(),
             self.introspection_tx.clone(),
-            read_holds_tx.clone(),
-            read_holds_rx,
         );
 
-        let instance = InstanceState::new(client, collections, read_holds_tx);
+        let instance = InstanceState::new(client, collections);
         self.instances.insert(id, instance);
 
         self.instance_workload_classes
@@ -608,7 +604,7 @@ where
     /// Panics if the identified `instance` still has active replicas.
     pub fn drop_instance(&mut self, id: ComputeInstanceId) {
         if let Some(instance) = self.instances.remove(&id) {
-            instance.call(|i| i.check_empty());
+            instance.call(|i| i.shutdown());
         }
 
         self.instance_workload_classes
@@ -1103,58 +1099,52 @@ struct InstanceState<T: ComputeControllerTimestamp> {
     client: instance::Client<T>,
     replicas: BTreeSet<ReplicaId>,
     collections: BTreeMap<GlobalId, Collection<T>>,
-    /// Sender for updates to collection read holds.
-    ///
-    /// Copies of this sender are given to [`ReadHold`]s that are created in
-    /// [`InstanceState::acquire_read_hold`].
-    read_holds_tx: mpsc::UnboundedSender<(GlobalId, ChangeBatch<T>)>,
 }
 
 impl<T: ComputeControllerTimestamp> InstanceState<T> {
-    fn new(
-        client: instance::Client<T>,
-        collections: BTreeMap<GlobalId, Collection<T>>,
-        read_holds_tx: mpsc::UnboundedSender<(GlobalId, ChangeBatch<T>)>,
-    ) -> Self {
+    fn new(client: instance::Client<T>, collections: BTreeMap<GlobalId, Collection<T>>) -> Self {
         Self {
             client,
             replicas: Default::default(),
             collections,
-            read_holds_tx,
         }
     }
 
     fn collection(&self, id: GlobalId) -> Result<&Collection<T>, CollectionMissing> {
         self.collections.get(&id).ok_or(CollectionMissing(id))
     }
 
-    pub fn call<F>(&self, f: F)
+    fn call<F>(&self, f: F)
     where
         F: FnOnce(&mut Instance<T>) + Send + 'static,
     {
         let otel_ctx = OpenTelemetryContext::obtain();
-        self.client.send(Box::new(move |instance| {
-            let _span = debug_span!("instance::call").entered();
-            otel_ctx.attach_as_parent();
-
-            f(instance)
-        }));
+        self.client
+            .send(Box::new(move |instance| {
+                let _span = debug_span!("instance::call").entered();
+                otel_ctx.attach_as_parent();
+
+                f(instance)
+            }))
+            .expect("instance not dropped");
     }
 
-    pub async fn call_sync<F, R>(&self, f: F) -> R
+    async fn call_sync<F, R>(&self, f: F) -> R
     where
         F: FnOnce(&mut Instance<T>) -> R + Send + 'static,
         R: Send + 'static,
     {
         let (tx, rx) = oneshot::channel();
         let otel_ctx = OpenTelemetryContext::obtain();
-        self.client.send(Box::new(move |instance| {
-            let _span = debug_span!("instance::call_sync").entered();
-            otel_ctx.attach_as_parent();
+        self.client
+            .send(Box::new(move |instance| {
+                let _span = debug_span!("instance::call_sync").entered();
+                otel_ctx.attach_as_parent();
 
-            let result = f(instance);
-            let _ = tx.send(result);
-        }));
+                let result = f(instance);
+                let _ = tx.send(result);
+            }))
+            .expect("instance not dropped");
 
         rx.await.expect("instance not dropped")
     }
@@ -1177,7 +1167,7 @@ impl<T: ComputeControllerTimestamp> InstanceState<T> {
             since
         });
 
-        let hold = ReadHold::new(id, since, self.read_holds_tx.clone());
+        let hold = ReadHold::new(id, since, self.client.read_hold_tx());
         Ok(hold)
     }
 
@@ -1196,7 +1186,6 @@ impl<T: ComputeControllerTimestamp> InstanceState<T> {
             client: _,
             replicas,
             collections,
-            read_holds_tx: _,
         } = self;
 
         let instance = self.call_sync(|i| i.dump()).await?;