Feature/graceful reconfig cancel

[jubrad]

Motivation

Adds cancelation to alter cluster for graceful reconfiguration.

This PR is stacked on the following PRs

• Feature/graceful reconfig v1 (wait for <duration>) #27904

Tips for reviewer

Checklist

• This PR has adequate test coverage / QA involvement has been duly considered. (trigger-ci for additional test/nightly runs)
• This PR has an associated up-to-date design doc, is a design doc (template), or is sufficiently small to not require a design.
• If this PR evolves an existing $T ⇔ Proto$T mapping (possibly in a backwards-incompatible way), then it is tagged with a T-proto label.
• 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).
• This PR includes the following user-facing behavior changes:

[jubrad]

I'd guess there's a much cleaner way of doing this rust, I gave it a couple of failed iterations and moved on.

[jkosh44]

Either

        let reconfiguring_clusters = self
            .active_conns
            .get_mut(conn_id)
            .expect("must exist for active session")
            .pending_cluster_alters
            .drain(..)
            .collect();
        self.drop_reconfiguration_replicas(reconfiguring_clusters)
            .await;

or

        let reconfiguring_clusters = std::mem::take(self
            .active_conns
            .get_mut(conn_id)
            .expect("must exist for active session")
            .pending_cluster_alters);
        self.drop_reconfiguration_replicas(reconfiguring_clusters)
            .await;

would probably work.

[jkosh44]

Why do we do a single transaction per replica instead of a single transaction with all replicas?

[jubrad]

confusingly, if I flatten the above Vec of Vecs and do a single transaction there are a lot of test failures that seem completely unrelated. For example, the following test fails ./mzcompose --find testdrive run default blue-green.td as well as cargo test.

Full test run:
https://buildkite.com/materialize/test/builds/88225#01914d7d-df74-4eba-ba28-33b389a503e0

I can't imagine a single connection will have more than cluster alter going at any given time, especially given that the execution is blocking. I'm pretty ok with leaving this a single transaction per cluster, but I am a bit confused and curious about why a single transaction for all replicas has this impact.

[jkosh44]

Either

        let reconfiguring_clusters = self
            .active_conns
            .get_mut(conn_id)
            .expect("must exist for active session")
            .pending_cluster_alters
            .drain(..)
            .collect();
        self.drop_reconfiguration_replicas(reconfiguring_clusters)
            .await;

or

        let reconfiguring_clusters = std::mem::take(self
            .active_conns
            .get_mut(conn_id)
            .expect("must exist for active session")
            .pending_cluster_alters);
        self.drop_reconfiguration_replicas(reconfiguring_clusters)
            .await;

would probably work.

[jkosh44]

LGTM