adapter: don't block the Coordinator on cluster status events (#31070)

While investigating some spikes in p99 message duration in the
Coordinator, I noticed that we currently block on a builtin table write
when receiving a status update for a cluster.

This PR refactors the code so we wait for the builtin write to complete
in a `tokio::task`, and then notify sessions.

### Motivation

Improve P99 latencies for Coordinator messages

### Checklist

- [x] This PR has adequate test coverage / QA involvement has been duly
considered. ([trigger-ci for additional test/nightly
runs](https://trigger-ci.dev.materialize.com/))
- [x] This PR has an associated up-to-date [design
doc](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/README.md),
is a design doc
([template](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/00000000_template.md)),
or is sufficiently small to not require a design.
  <!-- Reference the design in the description. -->
- [x] If this PR evolves [an existing `$T ⇔ Proto$T`
mapping](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/command-and-response-binary-encoding.md)
(possibly in a backwards-incompatible way), then it is tagged with a
`T-proto` label.
- [x] If this PR will require changes to cloud orchestration or tests,
there is a companion cloud PR to account for those changes that is
tagged with the release-blocker label
([example](MaterializeInc/cloud#5021)).
<!-- Ask in #team-cloud on Slack if you need help preparing the cloud
PR. -->
- [x] If this PR includes major [user-facing behavior
changes](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/guide-changes.md#what-changes-require-a-release-note),
I have pinged the relevant PM to schedule a changelog post.

src/adapter/src/coord.rs

@@ -3477,12 +3477,33 @@ impl Coordinator {
     }
 
     /// Publishes a notice message to all sessions.
+    ///
+    /// TODO(parkmycar): This code is dead, but is a nice parallel to [`Coordinator::broadcast_notice_tx`]
+    /// so we keep it around.
+    #[allow(dead_code)]
     pub(crate) fn broadcast_notice(&self, notice: AdapterNotice) {
         for meta in self.active_conns.values() {
             let _ = meta.notice_tx.send(notice.clone());
         }
     }
 
+    /// Returns a closure that will publish a notice to all sessions that were active at the time
+    /// this method was called.
+    pub(crate) fn broadcast_notice_tx(
+        &self,
+    ) -> Box<dyn FnOnce(AdapterNotice) -> () + Send + 'static> {
+        let senders: Vec<_> = self
+            .active_conns
+            .values()
+            .map(|meta| meta.notice_tx.clone())
+            .collect();
+        Box::new(move |notice| {
+            for tx in senders {
+                let _ = tx.send(notice.clone());
+            }
+        })
+    }
+
     pub(crate) fn active_conns(&self) -> &BTreeMap<ConnectionId, ConnMeta> {
         &self.active_conns
     }

src/adapter/src/coord/message_handler.rs

@@ -756,13 +756,8 @@ impl Coordinator {
             );
 
             let builtin_table_updates = vec![builtin_table_retraction, builtin_table_addition];
-
-            self.builtin_table_update()
-                .execute(builtin_table_updates)
-                .await
-                .0
-                .instrument(info_span!("coord::message_cluster_event::table_updates"))
-                .await;
+            // Returns a Future that completes when the Builtin Table write is completed.
+            let builtin_table_completion = self.builtin_table_update().defer(builtin_table_updates);
 
             let cluster = self.catalog().get_cluster(event.cluster_id);
             let replica = cluster.replica(event.replica_id).expect("Replica exists");
@@ -771,12 +766,24 @@ impl Coordinator {
                 .get_cluster_replica_status(event.cluster_id, event.replica_id);
 
             if old_replica_status != new_replica_status {
-                self.broadcast_notice(AdapterNotice::ClusterReplicaStatusChanged {
+                let notifier = self.broadcast_notice_tx();
+                let notice = AdapterNotice::ClusterReplicaStatusChanged {
                     cluster: cluster.name.clone(),
                     replica: replica.name.clone(),
                     status: new_replica_status,
                     time: event.time,
-                });
+                };
+                // In a separate task, so we don't block the Coordinator, wait for the builtin
+                // table update to complete, and then notify active sessions.
+                mz_ore::task::spawn(
+                    || format!("cluster_event-{}-{}", event.cluster_id, event.replica_id),
+                    async move {
+                        // Wait for the builtin table updates to complete.
+                        builtin_table_completion.await;
+                        // Notify all sessions that were active at the time the cluster status changed.
+                        (notifier)(notice)
+                    },
+                );
             }
         }
     }