platform-checks: fix silently ignored flags, replica masking, state leaks

misc/python/materialize/checks/executors.py

@@ -26,10 +26,13 @@ class Executor:
     # scenarios which are still in development and not available a few versions
     # back already.
     current_mz_version: MzVersion
-    # All the system settings we have already set in previous Mz versions. No
-    # need to set them again in a future version since they should be
-    # persisted.
-    system_settings: set[str] = set()
+
+    def __init__(self) -> None:
+        # All the system settings we have already set in previous Mz versions.
+        # No need to set them again in a future version since they should be
+        # persisted. Per instance so that a fresh environment (and its fresh
+        # executor) starts with a clean slate.
+        self.system_settings: set[str] = set()
 
     def testdrive(
         self, input: str, caller: Traceback | None = None, mz_service: str | None = None
@@ -55,6 +58,7 @@ def join(self, handle: Any) -> None:
 
 class MzcomposeExecutor(Executor):
     def __init__(self, composition: Composition) -> None:
+        super().__init__()
         self.composition = composition
 
     def mzcompose_composition(self) -> Composition:
@@ -85,7 +89,7 @@ def run_pyaction(
 
 class MzcomposeExecutorParallel(MzcomposeExecutor):
     def __init__(self, composition: Composition) -> None:
-        self.composition = composition
+        super().__init__(composition)
         self.exception: BaseException | None = None
 
     def testdrive(
@@ -148,6 +152,7 @@ def join(self, handle: Any) -> None:
 
 class CloudtestExecutor(Executor):
     def __init__(self, application: MaterializeApplication, version: MzVersion) -> None:
+        super().__init__()
         self.application = application
         self.seed = random.getrandbits(32)
         self.current_mz_version = version

misc/python/materialize/checks/mzcompose_actions.py

@@ -100,6 +100,13 @@ def execute(self, e: Executor) -> None:
             self.tag or MzVersion.parse_cargo(), config_name
         )
 
+        # Scenario-level parameters come from the command line
+        # (--system-param) and win over action-specific ones.
+        additional_system_parameter_defaults = {
+            **self.additional_system_parameter_defaults,
+            **self.scenario.additional_system_parameter_defaults,
+        }
+
         mz = Materialized(
             name=self.mz_service,
             image=image,
@@ -111,7 +118,7 @@ def execute(self, e: Executor) -> None:
             blob_store_is_azure=self.scenario.features.azurite_enabled(),
             environment_extra=self.environment_extra,
             system_parameter_defaults=self.system_parameter_defaults,
-            additional_system_parameter_defaults=self.additional_system_parameter_defaults,
+            additional_system_parameter_defaults=additional_system_parameter_defaults,
             system_parameter_version=self.system_parameter_version,
             soft_assertions=self.soft_assertions,
             sanity_restart=False,
@@ -121,7 +128,7 @@ def execute(self, e: Executor) -> None:
             restart=self.restart,
             force_migrations=self.force_migrations,
             publish=self.publish,
-            default_replication_factor=2,
+            default_replication_factor=self.scenario.default_replication_factor,
             support_external_clusterd=True,
             builtin_system_cluster_replication_factor=self.builtin_system_cluster_replication_factor,
             builtin_probe_cluster_replication_factor=self.builtin_probe_cluster_replication_factor,
@@ -214,10 +221,13 @@ def execute(self, e: Executor) -> None:
             """)
 
         self.handle = e.testdrive(input=input, mz_service=self.mz_service)
-        e.system_settings.update(system_settings)
+        self.applied_settings = system_settings
 
     def join(self, e: Executor) -> None:
         e.join(self.handle)
+        # Only record the settings as applied once the testdrive fragment that
+        # applies them has actually succeeded.
+        e.system_settings.update(self.applied_settings)
 
 
 class SetupSqlServerTesting(MzcomposeAction):
@@ -303,10 +313,23 @@ def execute(self, e: Executor) -> None:
             port=6877,
             user="mz_system",
         )
+
+        # Drop all existing replicas so that quickstart runs solely on
+        # clusterd_compute_1. A leftover orchestrated replica would keep
+        # serving queries and mask failures of the unorchestrated one.
+        replica_names = [row[0] for row in c.sql_query("""
+                SELECT mz_cluster_replicas.name
+                FROM mz_cluster_replicas
+                JOIN mz_clusters ON mz_cluster_replicas.cluster_id = mz_clusters.id
+                WHERE mz_clusters.name = 'quickstart'
+                """)]
+        drop_replicas = "\n".join(
+            f"DROP CLUSTER REPLICA quickstart.{name};" for name in replica_names
+        )
         c.sql(
-            """
+            f"""
             ALTER CLUSTER quickstart SET (MANAGED = false);
-            DROP CLUSTER REPLICA quickstart.r1;
+            {drop_replicas}
             CREATE CLUSTER REPLICA quickstart.r1
                 STORAGECTL ADDRESSES ['clusterd_compute_1:2100'],
                 STORAGE ADDRESSES ['clusterd_compute_1:2103'],

misc/python/materialize/checks/scenarios.py

@@ -9,6 +9,7 @@
 
 from dataclasses import dataclass
 from random import Random
+from typing import Any
 
 from materialize.checks.actions import Action, Initialize, Manipulate, Validate
 from materialize.checks.checks import Check
@@ -53,10 +54,20 @@ def __init__(
         executor: Executor,
         features: Features,
         seed: str | None = None,
+        additional_system_parameter_defaults: dict[str, str] | None = None,
+        default_replication_factor: int = 2,
     ) -> None:
-        self._checks = checks
+        # Copy the list since checks() shuffles it in place.
+        self._checks = list(checks)
         self.executor = executor
         self.features = features
+        # Applied by StartMz on top of any action-specific parameters so that
+        # --system-param and --default-replication-factor reach every Mz
+        # instance started by a scenario.
+        self.additional_system_parameter_defaults = (
+            additional_system_parameter_defaults or {}
+        )
+        self.default_replication_factor = default_replication_factor
         self.rng = None if seed is None else Random(seed)
         self._base_version = MzVersion.parse_cargo()
 
@@ -306,8 +317,9 @@ def __init__(
         features: Features,
         seed: str | None,
         change_entries: list[SystemVarChangeEntry],
+        **kwargs: Any,
     ):
-        super().__init__(checks, executor, features, seed)
+        super().__init__(checks, executor, features, seed, **kwargs)
         self.change_entries = change_entries
 
     def actions(self) -> list[Action]:

misc/python/materialize/checks/scenarios_backup_restore.py

@@ -26,14 +26,18 @@ class BackupAndRestoreAfterManipulate(Scenario):
     """
 
     def actions(self) -> list[Action]:
+        # Restart Mz through StartMz instead of Restore(restart_mz=True) so
+        # that the restarted instance uses the same service definition as the
+        # one it was originally started with.
         return [
             StartMz(self),
             Initialize(self),
             Manipulate(self, phase=1),
             Manipulate(self, phase=2),
             Backup(),
             KillMz(),
-            Restore(),
+            Restore(restart_mz=False),
+            StartMz(self),
             Validate(self),
         ]
 
@@ -48,7 +52,8 @@ def actions(self) -> list[Action]:
             Manipulate(self, phase=1),
             Backup(),
             KillMz(),
-            Restore(),
+            Restore(restart_mz=False),
+            StartMz(self),
             Manipulate(self, phase=2),
             Validate(self),
         ]
@@ -86,18 +91,22 @@ def actions(self) -> list[Action]:
             Initialize(self),
             Backup(),
             KillMz(),
-            Restore(),
+            Restore(restart_mz=False),
+            StartMz(self),
             Manipulate(self, phase=1),
             Backup(),
             KillMz(),
-            Restore(),
+            Restore(restart_mz=False),
+            StartMz(self),
             Manipulate(self, phase=2),
             Backup(),
             KillMz(),
-            Restore(),
+            Restore(restart_mz=False),
+            StartMz(self),
             Validate(self),
             Backup(),
             KillMz(),
-            Restore(),
+            Restore(restart_mz=False),
+            StartMz(self),
             Validate(self),
         ]

misc/python/materialize/checks/scenarios_upgrade.py

@@ -9,6 +9,7 @@
 
 
 from dataclasses import dataclass
+from typing import Any
 
 from materialize.checks.actions import Action, Initialize, Manipulate, Sleep, Validate
 from materialize.checks.checks import Check
@@ -100,11 +101,12 @@ def __init__(
         executor: Executor,
         features: Features,
         seed: str | None = None,
+        **kwargs: Any,
     ):
         self._self_managed_versions = get_self_managed_versions(
             max_version=MzVersion.parse_cargo()
         )
-        super().__init__(checks, executor, features, seed)
+        super().__init__(checks, executor, features, seed, **kwargs)
 
     def base_version(self) -> MzVersion:
         return self._self_managed_versions[-1]
@@ -146,11 +148,12 @@ def __init__(
         executor: Executor,
         features: Features,
         seed: str | None = None,
+        **kwargs: Any,
     ):
         self._self_managed_versions = get_self_managed_versions(
             max_version=MzVersion.parse_cargo()
         )
-        super().__init__(checks, executor, features, seed)
+        super().__init__(checks, executor, features, seed, **kwargs)
 
     def base_version(self) -> MzVersion:
         return self._self_managed_versions[-2]
@@ -273,9 +276,10 @@ def __init__(
         executor: Executor,
         features: Features,
         seed: str | None = None,
+        **kwargs: Any,
     ):
         self.minor_versions = get_minor_versions()
-        super().__init__(checks, executor, features, seed)
+        super().__init__(checks, executor, features, seed, **kwargs)
 
     def base_version(self) -> MzVersion:
         return self.minor_versions[3]
@@ -325,15 +329,6 @@ def actions(self) -> list[Action]:
 class UpgradeV80Migration(Scenario):
     """Test upgrade v26.17.1 (catalog 80) -> v26.18.0 (catalog 81, buggy v80_to_v81) -> v26.24.0 (catalog 83, partial fix in v82_to_v83) -> current (catalog 84, full fix in v83_to_v84)"""
 
-    def __init__(
-        self,
-        checks: list[type[Check]],
-        executor: Executor,
-        features: Features,
-        seed: str | None = None,
-    ):
-        super().__init__(checks, executor, features, seed)
-
     def base_version(self) -> MzVersion:
         return MzVersion.parse_mz("v26.17.1")
 
@@ -677,9 +672,10 @@ def __init__(
         executor: Executor,
         features: Features,
         seed: str | None = None,
+        **kwargs: Any,
     ):
         self.self_managed_versions = get_compatible_upgrade_from_versions()
-        super().__init__(checks, executor, features, seed)
+        super().__init__(checks, executor, features, seed, **kwargs)
 
     def base_version(self) -> MzVersion:
         return self.self_managed_versions[0]
@@ -733,9 +729,10 @@ def __init__(
         executor: Executor,
         features: Features,
         seed: str | None = None,
+        **kwargs: Any,
     ):
         self.self_managed_versions = get_compatible_upgrade_from_versions()
-        super().__init__(checks, executor, features, seed)
+        super().__init__(checks, executor, features, seed, **kwargs)
 
     def base_version(self) -> MzVersion:
         return self.self_managed_versions[0]
@@ -799,9 +796,10 @@ def __init__(
         executor: Executor,
         features: Features,
         seed: str | None = None,
+        **kwargs: Any,
     ):
         self.self_managed_versions = get_compatible_upgrade_from_versions()
-        super().__init__(checks, executor, features, seed)
+        super().__init__(checks, executor, features, seed, **kwargs)
 
     def _generate_random_upgrade_path(
         self,
@@ -913,9 +911,10 @@ def __init__(
         executor: Executor,
         features: Features,
         seed: str | None = None,
+        **kwargs: Any,
     ):
         self.self_managed_versions = get_compatible_upgrade_from_versions()
-        super().__init__(checks, executor, features, seed)
+        super().__init__(checks, executor, features, seed, **kwargs)
 
     def base_version(self) -> MzVersion:
         return self.self_managed_versions[0]

misc/python/materialize/checks/scenarios_zero_downtime.py

@@ -8,6 +8,8 @@
 # by the Apache License, Version 2.0.
 
 
+from typing import Any
+
 from materialize.checks.actions import (
     Action,
     BumpVersion,
@@ -293,9 +295,10 @@ def __init__(
         executor: Executor,
         features: Features,
         seed: str | None = None,
+        **kwargs: Any,
     ):
         self.minor_versions = get_minor_versions()
-        super().__init__(checks, executor, features, seed)
+        super().__init__(checks, executor, features, seed, **kwargs)
 
     def base_version(self) -> MzVersion:
         return self.minor_versions[3]

test/cloudtest/test_upgrade.py

@@ -93,6 +93,9 @@ def test_upgrade(aws_region: str | None, log_filter: str | None, dev: bool) -> N
             AlterConnectionHost,
         }
     )
+    # Sort so that parallel jobs, each of which computes its own shard, agree
+    # on the shard split. Set iteration order differs between processes.
+    checks.sort(key=lambda ch: ch.__name__)
     checks = buildkite.shard_list(checks, lambda ch: ch.__name__)
     if buildkite.get_parallelism_index() != 0 or buildkite.get_parallelism_count() != 1:
         print(