test: cover graceful cluster reconfiguration with single-replica sources

misc/python/materialize/checks/all_checks/cluster.py

@@ -9,7 +9,9 @@
 from textwrap import dedent
 
 from materialize.checks.actions import Testdrive
-from materialize.checks.checks import Check
+from materialize.checks.checks import Check, externally_idempotent
+from materialize.checks.executors import Executor
+from materialize.mz_version import MzVersion
 
 
 class CreateCluster(Check):
@@ -131,6 +133,123 @@ def validate(self) -> Testdrive:
            """))
 
 
+@externally_idempotent(False)
+class AlterClusterGracefulReconfiguration(Check):
+    """Graceful reconfiguration (WAIT UNTIL READY) of a cluster hosting a
+    single-replica Postgres source. Readiness must not wait for the source,
+    which never hydrates on the target replicas, and the settled
+    reconfiguration must survive restarts and upgrades. Regression coverage
+    for SQL-530.
+    """
+
+    def _can_run(self, e: Executor) -> bool:
+        # Graceful reconfiguration of a cluster hosting a single-replica
+        # source wedges until its deadline rolls it back on versions without
+        # the SQL-530 fix, so no phase may run on an older version. The fix
+        # predates the v26.35.0-rc.1 cut, so every published v26.35 artifact
+        # has it and the -dev gate includes the release candidates.
+        return self.base_version >= MzVersion.parse_mz("v26.35.0-dev")
+
+    def initialize(self) -> Testdrive:
+        return Testdrive(dedent("""
+            $ postgres-execute connection=postgres://mz_system@${testdrive.materialize-internal-sql-addr}
+            ALTER SYSTEM SET enable_zero_downtime_cluster_reconfiguration = true
+
+            $ postgres-execute connection=postgres://postgres:postgres@postgres
+            CREATE USER graceful_reconfig WITH SUPERUSER PASSWORD 'postgres';
+            ALTER USER graceful_reconfig WITH replication;
+            DROP PUBLICATION IF EXISTS graceful_reconfig_pub;
+            DROP TABLE IF EXISTS graceful_reconfig_table;
+            CREATE TABLE graceful_reconfig_table (f1 INTEGER PRIMARY KEY);
+            ALTER TABLE graceful_reconfig_table REPLICA IDENTITY FULL;
+            INSERT INTO graceful_reconfig_table SELECT generate_series(1, 100);
+            CREATE PUBLICATION graceful_reconfig_pub FOR TABLE graceful_reconfig_table;
+
+            > CREATE CLUSTER graceful_reconfig_cluster (SIZE 'scale=1,workers=1', REPLICATION FACTOR 1)
+
+            > CREATE SECRET graceful_reconfig_pass AS 'postgres'
+
+            > CREATE CONNECTION graceful_reconfig_conn FOR POSTGRES
+              HOST 'postgres',
+              DATABASE postgres,
+              USER graceful_reconfig,
+              PASSWORD SECRET graceful_reconfig_pass
+
+            > CREATE SOURCE graceful_reconfig_source
+              IN CLUSTER graceful_reconfig_cluster
+              FROM POSTGRES CONNECTION graceful_reconfig_conn
+              (PUBLICATION 'graceful_reconfig_pub')
+
+            > CREATE TABLE graceful_reconfig_t FROM SOURCE graceful_reconfig_source (REFERENCE graceful_reconfig_table)
+
+            > SELECT count(*) FROM graceful_reconfig_t
+            100
+        """))
+
+    def manipulate(self) -> list[Testdrive]:
+        return [
+            Testdrive(dedent(s))
+            for s in [
+                """
+                $ set-sql-timeout duration=240s
+
+                > ALTER CLUSTER graceful_reconfig_cluster SET (SIZE 'scale=1,workers=2') WITH (WAIT UNTIL READY (TIMEOUT '300s', ON TIMEOUT ROLLBACK))
+
+                # A background ALTER returns immediately, so wait for the
+                # realized config to cut over to the target.
+                > SELECT size FROM mz_clusters WHERE name = 'graceful_reconfig_cluster'
+                "scale=1,workers=2"
+
+                # The source keeps ingesting on the promoted replica set.
+                $ postgres-execute connection=postgres://postgres:postgres@postgres
+                INSERT INTO graceful_reconfig_table SELECT generate_series(101, 200);
+
+                > SELECT count(*) FROM graceful_reconfig_t
+                200
+
+                $ set-sql-timeout duration=default
+                """,
+                """
+                $ set-sql-timeout duration=240s
+
+                > ALTER CLUSTER graceful_reconfig_cluster SET (SIZE 'scale=2,workers=2') WITH (WAIT UNTIL READY (TIMEOUT '300s', ON TIMEOUT ROLLBACK))
+
+                > SELECT size FROM mz_clusters WHERE name = 'graceful_reconfig_cluster'
+                "scale=2,workers=2"
+
+                $ postgres-execute connection=postgres://postgres:postgres@postgres
+                INSERT INTO graceful_reconfig_table SELECT generate_series(201, 300);
+
+                > SELECT count(*) FROM graceful_reconfig_t
+                300
+
+                $ set-sql-timeout duration=default
+                """,
+            ]
+        ]
+
+    def validate(self) -> Testdrive:
+        return Testdrive(dedent("""
+            $ set-sql-timeout duration=240s
+
+            > SELECT size FROM mz_clusters WHERE name = 'graceful_reconfig_cluster'
+            "scale=2,workers=2"
+
+            # The settled reconfiguration record is retained across restarts
+            # and upgrades.
+            > SELECT recon.status
+              FROM mz_internal.mz_cluster_reconfigurations recon
+              JOIN mz_clusters c ON c.id = recon.cluster_id
+              WHERE c.name = 'graceful_reconfig_cluster'
+            finalized
+
+            > SELECT count(*) FROM graceful_reconfig_t
+            300
+
+            $ set-sql-timeout duration=default
+        """))
+
+
 class DropCluster(Check):
     def manipulate(self) -> list[Testdrive]:
         return [

misc/python/materialize/parallel_workload/action.py

@@ -2133,8 +2133,12 @@ def errors_to_ignore(self, exe: Executor) -> list[str]:
             # `EXPERIMENTAL ARRANGEMENT COMPRESSION` is managed-only. The
             # tracked `managed` flag can lag reality (e.g. an in-flight CREATE
             # lost to a kill), so the option can land on a cluster that is
-            # unmanaged by the time the ALTER runs.
+            # unmanaged by the time the ALTER runs. Two code paths reject it:
+            # the planner and the sequencer's defense-in-depth check, each with
+            # its own message, and which one fires depends on the timing of the
+            # desync.
             "EXPERIMENTAL ARRANGEMENT COMPRESSION not supported for unmanaged clusters",
+            "Cannot change EXPERIMENTAL ARRANGEMENT COMPRESSION of unmanaged clusters",
         ] + super().errors_to_ignore(exe)
 
     def run(self, exe: Executor) -> bool:
@@ -2181,10 +2185,20 @@ def run(self, exe: Executor) -> bool:
         conn = None
 
         try:
-            conn = self.create_system_connection(exe)
             if cluster is not None:
-                self.set_cluster_compression(conn, cluster)
+                # Serialize with ReconfigureClusterAction: any managed-to-
+                # managed ALTER CLUSTER folds onto an in-flight graceful
+                # reconfiguration record and, lacking a WAIT clause, replaces
+                # its deadline with the 24h default (SQL-568), which strands
+                # the reconfiguration that action is polling on.
+                # TODO: Reenable running without the lock when SQL-568 is fixed
+                with cluster.lock:
+                    if cluster not in exe.db.clusters:
+                        return False
+                    conn = self.create_system_connection(exe)
+                    self.set_cluster_compression(conn, cluster)
             else:
+                conn = self.create_system_connection(exe)
                 self.flip_flag(conn, flag_name, flag_value)
                 exe.db.flags[flag_name] = flag_value
             return True
@@ -2577,6 +2591,91 @@ def run(self, exe: Executor) -> bool:
         return True
 
 
+class ReconfigureClusterAction(Action):
+    """Gracefully resize a random managed cluster and assert convergence.
+
+    Regression coverage for SQL-530: readiness must not wait for
+    single-replica sources (Postgres, MySQL, SQL Server) hosted on the
+    cluster, which never hydrate on the reconfiguration's target replicas
+    and used to wedge the reconfiguration until its deadline rolled it back.
+    """
+
+    def applicable(self, exe: Executor) -> bool:
+        # Convergence cannot be asserted while environmentd is killed,
+        # restored from backup, or fenced out by a 0dt deploy.
+        return exe.db.scenario not in (
+            Scenario.Kill,
+            Scenario.BackupRestore,
+            Scenario.ZeroDowntimeDeploy,
+        )
+
+    def run(self, exe: Executor) -> bool:
+        with exe.db.lock:
+            managed_clusters = [c for c in exe.db.clusters if c.managed]
+            if not managed_clusters:
+                return False
+            cluster = self.rng.choice(managed_clusters)
+        # Hold the cluster lock for the whole reconfiguration so no concurrent
+        # rename, swap, or drop invalidates the polled name and no concurrent
+        # resize folds another size into the in-flight record.
+        with cluster.lock:
+            if cluster not in exe.db.clusters:
+                return False
+            new_size = self.rng.choice(
+                [
+                    s
+                    for s in [
+                        "scale=1,workers=1",
+                        "scale=1,workers=4",
+                        "scale=2,workers=2",
+                    ]
+                    if s != cluster.size
+                ]
+            )
+            alter_started = time.time()
+            exe.execute(
+                f"ALTER CLUSTER {cluster} SET (SIZE '{new_size}') WITH (WAIT UNTIL READY (TIMEOUT '120s', ON TIMEOUT ROLLBACK))",
+                http=Http.RANDOM,
+            )
+            # With background ALTER CLUSTER the statement returns immediately
+            # and the controller converges the durable record, so poll the
+            # realized config. A rollback at the deadline (status "timed-out")
+            # is a legitimate outcome: readiness can be genuinely unreachable,
+            # e.g. when another worker put a REFRESH EVERY materialized view
+            # with a far-future refresh on the cluster, which cannot hydrate
+            # on the target replicas before its refresh time. What must never
+            # happen is the record staying "in-progress" past its deadline,
+            # which is how a wedged readiness check (SQL-530) manifests. Only
+            # trust a terminal status once the ALTER's own deadline has
+            # passed: builtin table reads can lag the catalog, so before that
+            # a terminal status can only be a stale record from an earlier
+            # reconfiguration of this cluster.
+            name_literal = cluster.name().replace("'", "''")
+            query = (
+                "SELECT c.size, r.status FROM mz_clusters c "
+                "LEFT JOIN mz_internal.mz_cluster_reconfigurations r ON r.cluster_id = c.id "
+                f"WHERE c.name = '{name_literal}'"
+            )
+            status = None
+            deadline = time.time() + 180
+            while time.time() < deadline:
+                exe.execute(query)
+                size, status = exe.cur.fetchall()[0]
+                if size == new_size:
+                    cluster.size = new_size
+                    return True
+                if time.time() > alter_started + 120 and status in (
+                    "timed-out",
+                    "resource-exhausted",
+                ):
+                    return True
+                time.sleep(1)
+            raise ValueError(
+                f"Graceful reconfiguration of cluster {cluster} to size {new_size} "
+                f"did not complete (reconfiguration status: {status})"
+            )
+
+
 class GrantPrivilegesAction(Action):
     def run(self, exe: Executor) -> bool:
         with exe.db.lock:
@@ -2743,7 +2842,16 @@ def pg_port() -> int:
                 # Reapply the session settings from Worker.run, they don't
                 # survive the reconnect.
                 cur.execute("SET auto_route_catalog_queries TO false")
-                cur.execute("SELECT pg_backend_pid()")
+                try:
+                    cur.execute("SELECT pg_backend_pid()")
+                except Exception as e:
+                    # FlipFlags may have poisoned the global default cluster to
+                    # `dont_exist`. Reconnecting can't fix a missing default, so
+                    # pin the session to a live cluster instead of wedging.
+                    if "unknown cluster 'dont_exist'" not in str(e):
+                        raise
+                    self._pin_default_cluster(exe, conn, cur)
+                    cur.execute("SELECT pg_backend_pid()")
                 if not exe.use_ws:
                     exe.pg_pid = cur.fetchall()[0][0]
             except Exception as e:
@@ -2767,6 +2875,27 @@ def pg_port() -> int:
                 break
         return True
 
+    def _pin_default_cluster(
+        self, exe: Executor, conn: Connection, cur: psycopg.Cursor
+    ) -> None:
+        """Pin the reconnected session onto `quickstart` (recreated at setup,
+        never dropped) after the global default was poisoned to `dont_exist`.
+        The WS session inherits the same poisoned default and must be pinned
+        too. `SET cluster` is rejected inside a transaction, so clear the
+        aborted probe transaction and run it in autocommit mode."""
+        conn.rollback()
+        prev_autocommit = conn.autocommit
+        conn.autocommit = True
+        cur.execute("SET cluster = quickstart")
+        conn.autocommit = prev_autocommit
+        if exe.use_ws:
+            # Best effort: on failure the worker just fails WS queries until
+            # FlipFlags resets the default, still better than wedging.
+            try:
+                exe.ws_query("SET cluster = quickstart;")
+            except QueryError:
+                pass
+
 
 class CancelAction(Action):
     workers: list["Worker"]
@@ -3737,6 +3866,7 @@ def __init__(
         (SwapClusterAction, 10),
         (CreateClusterReplicaAction, 2),
         (DropClusterReplicaAction, 2),
+        (ReconfigureClusterAction, 1),
         (SetClusterAction, 1),
         (CreateWebhookSourceAction, 2),
         (DropWebhookSourceAction, 2),

misc/python/materialize/parallel_workload/parallel_workload.py

@@ -141,6 +141,10 @@ def run(
         system_exe.execute("ALTER SYSTEM SET max_sql_server_connections = 1000000")
         system_exe.execute("ALTER SYSTEM SET max_kafka_connections = 1000000")
         system_exe.execute("ALTER SYSTEM SET idle_in_transaction_session_timeout = 0")
+        # Gates the WITH (WAIT ...) clause used by ReconfigureClusterAction.
+        system_exe.execute(
+            "ALTER SYSTEM SET enable_zero_downtime_cluster_reconfiguration = true"
+        )
         # Most queries should not fail because of privileges
         for object_type in [
             "TABLES",

misc/python/materialize/parallel_workload/worker.py

@@ -126,15 +126,17 @@ def run(
                 self.run_action(
                     ReconnectAction(self.rng, self.composition, random_role=False)
                 )
-                if self.exe.reconnect_next or self.exe.rollback_next:
-                    # Reconnecting failed with an ignored error. Always retry
-                    # the reconnect, never fall through to the action: the
-                    # old session may hold an aborted transaction that fails
-                    # all statements.
-                    self.exe.reconnect_next = True
+                if self.exe.reconnect_next:
+                    # The connection couldn't be re-established, retry it.
                     self.exe.rollback_next = False
                     time.sleep(1)
                     continue
+                if self.exe.rollback_next:
+                    # Connected, but a probe hit a non-connection error (e.g. a
+                    # poisoned default cluster). Retrying the reconnect can't fix
+                    # that, so let the loop's rollback path clear the aborted
+                    # transaction instead of wedging.
+                    continue
             self.run_action(action)
 
         self.exe.cur.connection.close()

test/cloudtest/setup

@@ -16,7 +16,7 @@ cd "$(dirname "$0")/../.."
 . misc/shlib/shlib.bash
 . test/cloudtest/config.bash
 
-if ! kind get clusters | grep -q "$K8S_CLUSTER_NAME"; then
+create_cluster() {
     CLUSTER_YAML=${CLOUDTEST_CLUSTER_DEFINITION_FILE:-"misc/kind/cluster.yaml"}
     echo "Using cluster definition specified in ${CLUSTER_YAML}"
 
@@ -25,8 +25,19 @@ if ! kind get clusters | grep -q "$K8S_CLUSTER_NAME"; then
     run kind get kubeconfig --name="$K8S_CLUSTER_NAME" > "$KUBECONFIG"
 
     run kubectl --context="$K8S_CONTEXT" taint nodes --selector=environmentd=true "environmentd=true:NoSchedule"
-else
-    run kind get kubeconfig --name="$K8S_CLUSTER_NAME" > "$KUBECONFIG"
+}
+
+if ! kind get clusters | grep -q "$K8S_CLUSTER_NAME"; then
+    create_cluster
+elif ! kind get kubeconfig --name="$K8S_CLUSTER_NAME" > "$KUBECONFIG"; then
+    # The cluster is listed but unusable: kind lists a cluster as long as any
+    # node container exists, but an interrupted cleanup of a previous CI job
+    # on this agent can leave only some of the node containers behind (e.g.
+    # missing control plane, or nodes vanishing mid-inspect). Recreate it
+    # from scratch.
+    echo "kind cluster $K8S_CLUSTER_NAME is broken, recreating it"
+    run kind delete cluster --name="$K8S_CLUSTER_NAME"
+    create_cluster
 fi
 
 for f in misc/kind/configmaps/*; do