From 8ce5f9b2bbe9c8891aba3e580fe7895a049437f5 Mon Sep 17 00:00:00 2001 From: Dennis Felsing Date: Fri, 24 Jul 2026 06:33:12 +0000 Subject: [PATCH 1/2] ci: Fix nightly failures (2026-07-24) Based on https://buildkite.com/materialize/nightly/builds/17526 & https://buildkite.com/materialize/nightly/builds/17523 --- .../materialize/parallel_workload/action.py | 38 ++++++++++++++++++- .../materialize/parallel_workload/worker.py | 14 ++++--- test/sqlancer/Dockerfile | 2 +- 3 files changed, 45 insertions(+), 9 deletions(-) diff --git a/misc/python/materialize/parallel_workload/action.py b/misc/python/materialize/parallel_workload/action.py index 6ee0a98ee9a42..05a3d5b9ca820 100644 --- a/misc/python/materialize/parallel_workload/action.py +++ b/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: @@ -2743,7 +2747,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 +2780,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"] diff --git a/misc/python/materialize/parallel_workload/worker.py b/misc/python/materialize/parallel_workload/worker.py index f6c6739c6c53e..b37e9406e3e00 100644 --- a/misc/python/materialize/parallel_workload/worker.py +++ b/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() diff --git a/test/sqlancer/Dockerfile b/test/sqlancer/Dockerfile index de23e5c4d6eed..ef375556a10fb 100644 --- a/test/sqlancer/Dockerfile +++ b/test/sqlancer/Dockerfile @@ -26,7 +26,7 @@ RUN apt-get update && TZ=UTC DEBIAN_FRONTEND=noninteractive apt-get install -y - # ggevay: But in the meantime there have been more changes to our fork, see the `main` branch there. RUN git clone https://github.com/MaterializeInc/sqlancer \ && cd sqlancer \ - && git checkout 2705f070891104d380666aa070ccfa40f859849c \ + && git checkout 4e116a5bb6c57ac46c2df5f68dd35c3534fcc3f4 \ && rm -rf .git \ && mvn package -DskipTests # RUN git clone --depth=1 --single-branch https://github.com/sqlancer/sqlancer \ From 8b6c462e2355187648f89c29efd138767623e4e2 Mon Sep 17 00:00:00 2001 From: Dennis Felsing Date: Fri, 24 Jul 2026 08:10:35 +0000 Subject: [PATCH 2/2] test: cover graceful cluster reconfiguration with single-replica sources ALTER CLUSTER ... WITH (WAIT UNTIL READY) deadlocking on clusters that host a single-replica source (SQL-530, fixed in #37740) shipped because no test combined a readiness-gated reconfiguration with a Postgres, MySQL, or SQL Server source. Close the gap from three sides: * parallel-workload: a new ReconfigureClusterAction gracefully resizes a random managed cluster with ON TIMEOUT ROLLBACK and fails the run if the realized config never cuts over, so a wedged readiness check surfaces as a test failure instead of a silent rollback. Sources land on random clusters, so resizes regularly hit clusters hosting Postgres sources. * platform-checks: a new AlterClusterGracefulReconfiguration check reconfigures a cluster hosting a Postgres source in both manipulate phases and validates that the realized config, the finalized reconfiguration record, and ingestion survive restarts and upgrades. Gated on base version v26.35.0-dev: the fix predates the v26.35.0-rc.1 cut, and older versions wedge on the reconfiguration itself, so no phase may run there. * cloudtest: test_zero_downtime_reconfiguration now resizes a cluster hosting a Postgres source and asserts the reconfiguration finalizes and ingestion continues across cut-over, covering the k8s path where target replica pods must come online before cut-over. Co-Authored-By: Claude Fable 5 --- .../materialize/checks/all_checks/cluster.py | 121 +++++++++++++++++- .../materialize/parallel_workload/action.py | 100 ++++++++++++++- .../parallel_workload/parallel_workload.py | 4 + test/cloudtest/setup | 17 ++- test/cloudtest/test_managed_cluster.py | 70 ++++++++++ 5 files changed, 306 insertions(+), 6 deletions(-) diff --git a/misc/python/materialize/checks/all_checks/cluster.py b/misc/python/materialize/checks/all_checks/cluster.py index ea841fabc081c..65a4bcb12ad1e 100644 --- a/misc/python/materialize/checks/all_checks/cluster.py +++ b/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 [ diff --git a/misc/python/materialize/parallel_workload/action.py b/misc/python/materialize/parallel_workload/action.py index 05a3d5b9ca820..88a8795a15103 100644 --- a/misc/python/materialize/parallel_workload/action.py +++ b/misc/python/materialize/parallel_workload/action.py @@ -2185,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 @@ -2581,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: @@ -3771,6 +3866,7 @@ def __init__( (SwapClusterAction, 10), (CreateClusterReplicaAction, 2), (DropClusterReplicaAction, 2), + (ReconfigureClusterAction, 1), (SetClusterAction, 1), (CreateWebhookSourceAction, 2), (DropWebhookSourceAction, 2), diff --git a/misc/python/materialize/parallel_workload/parallel_workload.py b/misc/python/materialize/parallel_workload/parallel_workload.py index 3bdecddac148f..6124fc5cdc639 100644 --- a/misc/python/materialize/parallel_workload/parallel_workload.py +++ b/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", diff --git a/test/cloudtest/setup b/test/cloudtest/setup index 5d9934664ec19..63a9e93c5d86f 100755 --- a/test/cloudtest/setup +++ b/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 diff --git a/test/cloudtest/test_managed_cluster.py b/test/cloudtest/test_managed_cluster.py index 84bf757e12169..46836c605d61a 100644 --- a/test/cloudtest/test_managed_cluster.py +++ b/test/cloudtest/test_managed_cluster.py @@ -511,3 +511,73 @@ def query_with_conn( """), no_reset=True, ) + + # A single-replica source (Postgres CDC) stays scheduled on the replica it + # already runs on and never hydrates on a reconfiguration's target + # replicas. Readiness must skip it and instead gate on the target replicas + # being online, so the resize completes and cuts over. Regression coverage + # for SQL-530, where this wedged until the deadline rolled the resize back. + mz.testdrive.run( + input=dedent(""" + $ postgres-execute connection=postgres://postgres:postgres@postgres + ALTER USER postgres WITH replication; + DROP SCHEMA IF EXISTS public CASCADE; + DROP PUBLICATION IF EXISTS mz_source; + CREATE SCHEMA public; + CREATE TABLE pg_t (f1 INTEGER PRIMARY KEY); + ALTER TABLE pg_t REPLICA IDENTITY FULL; + INSERT INTO pg_t VALUES (1), (2), (3); + CREATE PUBLICATION mz_source FOR TABLE pg_t; + + > CREATE CLUSTER source_reconfig (SIZE 'scale=1,workers=1', REPLICATION FACTOR 1) + + > CREATE SECRET pg_source_pass AS 'postgres' + + > CREATE CONNECTION pg_source_conn TO POSTGRES ( + HOST 'postgres', + DATABASE postgres, + USER postgres, + PASSWORD SECRET pg_source_pass + ) + + > CREATE SOURCE pg_source + IN CLUSTER source_reconfig + FROM POSTGRES CONNECTION pg_source_conn (PUBLICATION 'mz_source') + + > CREATE TABLE pg_t FROM SOURCE pg_source (REFERENCE pg_t) + + > SELECT * FROM pg_t + 1 + 2 + 3 + + > ALTER CLUSTER source_reconfig SET (SIZE 'scale=1,workers=2') WITH (WAIT UNTIL READY (TIMEOUT '300s', ON TIMEOUT ROLLBACK)) + + # The background ALTER returns immediately. The raised timeout + # covers the target replica pod's boot before the cut-over. + $ set-sql-timeout duration=120s + + > SELECT size FROM mz_clusters WHERE name = 'source_reconfig' + "scale=1,workers=2" + + > SELECT recon.status + FROM mz_internal.mz_cluster_reconfigurations recon + JOIN mz_clusters c ON c.id = recon.cluster_id + WHERE c.name = 'source_reconfig' + finalized + + $ set-sql-timeout duration=default + + # The source keeps serving and ingesting across the cut-over. + $ postgres-execute connection=postgres://postgres:postgres@postgres + INSERT INTO pg_t VALUES (4); + + > SELECT * FROM pg_t + 1 + 2 + 3 + 4 + """), + no_reset=True, + ) + assert_no_pending("source_reconfig")