test/launchdarkly: cover per-cluster and per-replica scoped flags

Extend the LaunchDarkly integration test with cluster-coherent and
replica-local cases. Adds rule-based targeting (clauses on context-kind
attributes) and boolean flags, then demonstrates:

* a cluster-coherent optimizer feature served `true` to builtin clusters
  (targeted by the `cluster` context's `is_builtin`) while user clusters
  fall through to `false`, observed via `mz_cluster_system_parameters`;
* `enable_lgalloc` served `false` only to `legacy` size-family replicas
  (targeted by the `replica` context's `replica_size_family`) while `cc`
  replicas keep the env-wide value, observed via
  `mz_replica_system_parameters`;
* durability of both across a restart with the sync loop disabled.

The LAUNCHDARKLY_KEY_MAP entries are `;`-separated (the CLI arg's value
delimiter). The LaunchDarkly project/environment keys are overridable via
LAUNCHDARKLY_PROJECT_KEY / LAUNCHDARKLY_ENVIRONMENT_KEY for running
against a non-CI project.

Verified end-to-end against a live LaunchDarkly test project.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: antiguru

test/launchdarkly/mzcompose.py

@@ -50,6 +50,39 @@
 LD_CONTEXT_KEY = DEFAULT_MZ_ENVIRONMENT_ID
 # A unique feature flag key to use for this test.
 LD_FEATURE_FLAG_KEY = f"ci-test-{BUILDKITE_JOB_ID}"
+# Unique feature flag keys for the scoped (per-cluster / per-replica) cases.
+LD_OPTIMIZER_FLAG_KEY = f"ci-test-optimizer-{BUILDKITE_JOB_ID}"
+LD_LGALLOC_FLAG_KEY = f"ci-test-lgalloc-{BUILDKITE_JOB_ID}"
+# A cluster-coherent (optimizer) parameter and a replica-local parameter, both
+# declared scoped in their definitions.
+OPTIMIZER_PARAM = "enable_eager_delta_joins"
+LGALLOC_PARAM = "enable_lgalloc"
+
+
+def context_rule(
+    context_kind: str, attribute: str, values: list[str], variation: int
+) -> dict[str, Any]:
+    """Builds a LaunchDarkly targeting rule that serves `variation` to contexts
+    of `context_kind` whose `attribute` is in `values`."""
+    return {
+        "variation": variation,
+        "clauses": [
+            {
+                "contextKind": context_kind,
+                "attribute": attribute,
+                "op": "in",
+                "values": values,
+                "negate": False,
+            }
+        ],
+    }
+
+
+# Boolean flag variations: index 0 is `false`, index 1 is `true`.
+BOOL_VARIATIONS: list[Any] = [
+    Variation(value=False, name="false"),
+    Variation(value=True, name="true"),
+]
 
 SERVICES = [
     Materialized(
@@ -79,8 +112,8 @@ def workflow_default(c: Composition) -> None:
         configuration=launchdarkly_api.Configuration(
             api_key=dict(ApiKey=LAUNCHDARKLY_API_TOKEN),
         ),
-        project_key="default",
-        environment_key="ci-cd",
+        project_key=environ.get("LAUNCHDARKLY_PROJECT_KEY", "default"),
+        environment_key=environ.get("LAUNCHDARKLY_ENVIRONMENT_KEY", "ci-cd"),
     )
 
     try:
@@ -229,6 +262,9 @@ def sys(command: str) -> None:
         # turned off).
         c.testdrive("\n".join(["> SHOW max_result_size", "1GB"]))
         c.stop("materialized")
+
+        # Exercise the scoped (per-cluster / per-replica) feature flags.
+        run_scoped_feature_flag_cases(c, ld_client)
     except launchdarkly_api.ApiException as e:
         raise UIError(dedent(f"""
                 Error when calling the Launch Darkly API.
@@ -242,6 +278,145 @@ def sys(command: str) -> None:
             pass  # ignore exceptions on cleanup
 
 
+def run_scoped_feature_flag_cases(
+    c: Composition, ld_client: "LaunchDarklyClient"
+) -> None:
+    """Demonstrates per-cluster (cluster-coherent) and per-replica
+    (replica-local) LaunchDarkly overrides, plus their durability across a
+    restart.
+
+    Cluster case: a cluster-coherent optimizer feature is served `true` to the
+    builtin clusters (e.g. `mz_catalog_server`) and `false` to user clusters,
+    targeted by the `cluster` context's `is_builtin` attribute.
+
+    Replica case: the replica-local `enable_lgalloc` flag is served `false` only
+    to replicas whose `replica_size_family` is `legacy`, while modern (`cc`)
+    replicas keep the environment-wide value, targeted by the `replica`
+    context's `replica_size_family` attribute.
+    """
+    # A materialized that syncs the scoped flags. The key map ties the
+    # cluster-coherent optimizer parameter and the replica-local lgalloc
+    # parameter to their LD flags, alongside the original max_result_size flag.
+    # MZ_LAUNCHDARKLY_KEY_MAP entries are `;`-separated (the CLI arg's value
+    # delimiter), not comma-separated.
+    key_map = ";".join(
+        [
+            f"max_result_size={LD_FEATURE_FLAG_KEY}",
+            f"{OPTIMIZER_PARAM}={LD_OPTIMIZER_FLAG_KEY}",
+            f"{LGALLOC_PARAM}={LD_LGALLOC_FLAG_KEY}",
+        ]
+    )
+    scoped_mz = Materialized(
+        environment_extra=[
+            f"MZ_LAUNCHDARKLY_SDK_KEY={LAUNCHDARKLY_SDK_KEY}",
+            f"MZ_LAUNCHDARKLY_KEY_MAP={key_map}",
+            "MZ_CONFIG_SYNC_LOOP_INTERVAL=1s",
+        ],
+        additional_system_parameter_defaults={
+            "log_filter": "mz_adapter::catalog=debug,mz_adapter::config=debug",
+        },
+        external_metadata_store=True,
+    )
+
+    try:
+        # Cluster-coherent optimizer flag: fallthrough serves `false`, a rule
+        # serves `true` to builtin clusters.
+        ld_client.create_flag(
+            LD_OPTIMIZER_FLAG_KEY,
+            tags=["ci-test"],
+            variations=BOOL_VARIATIONS,
+            off_variation=0,
+            on_variation=0,
+        )
+        ld_client.update_targeting(
+            LD_OPTIMIZER_FLAG_KEY,
+            on=True,
+            rules=[context_rule("cluster", "is_builtin", ["true"], 1)],
+        )
+
+        # Replica-local lgalloc flag: fallthrough serves `true` (the env-wide
+        # default), a rule serves `false` to the legacy size family.
+        ld_client.create_flag(
+            LD_LGALLOC_FLAG_KEY,
+            tags=["ci-test"],
+            variations=BOOL_VARIATIONS,
+            off_variation=0,
+            on_variation=1,
+        )
+        ld_client.update_targeting(
+            LD_LGALLOC_FLAG_KEY,
+            on=True,
+            rules=[context_rule("replica", "replica_size_family", ["legacy"], 0)],
+        )
+
+        with c.override(scoped_mz):
+            c.up("materialized")
+
+            # A user cluster with a legacy-family replica, to contrast with the
+            # builtin/`cc` defaults.
+            c.testdrive(
+                "\n".join(
+                    [
+                        "$ postgres-connect name=mz_system url=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr}",
+                        "$ postgres-execute connection=mz_system",
+                        "CREATE CLUSTER ld_legacy SIZE 'scale=1,workers=1,legacy'",
+                    ]
+                )
+            )
+
+            # Allow a few sync ticks to evaluate the new cluster/replica.
+            sleep(5)
+
+            # Cluster case: the catalog server gets the cluster-scoped `true`,
+            # while the user cluster falls through to `false`.
+            c.testdrive(
+                "\n".join(
+                    [
+                        f"> SELECT c.name, p.value FROM mz_internal.mz_cluster_system_parameters p JOIN mz_clusters c ON c.id = p.cluster_id WHERE p.name = '{OPTIMIZER_PARAM}' AND c.name IN ('mz_catalog_server', 'quickstart') ORDER BY c.name",
+                        "mz_catalog_server true",
+                        "quickstart false",
+                    ]
+                )
+            )
+
+            # Replica case: the legacy-family replica gets `false`, while the
+            # `cc`-family quickstart replica keeps the env-wide `true`.
+            c.testdrive(
+                "\n".join(
+                    [
+                        f"> SELECT cr.name, p.value FROM mz_internal.mz_replica_system_parameters p JOIN mz_cluster_replicas cr ON cr.id = p.replica_id JOIN mz_clusters c ON c.id = cr.cluster_id WHERE p.name = '{LGALLOC_PARAM}' AND c.name IN ('ld_legacy', 'quickstart') ORDER BY c.name",
+                        "r1 false",
+                        "r1 true",
+                    ]
+                )
+            )
+            c.stop("materialized")
+
+        # Durability: restart without the sync loop and assert the scoped
+        # values are restored from the durable cache.
+        with c.override(Materialized(external_metadata_store=True)):
+            c.up("materialized")
+            c.testdrive(
+                "\n".join(
+                    [
+                        f"> SELECT c.name, p.value FROM mz_internal.mz_cluster_system_parameters p JOIN mz_clusters c ON c.id = p.cluster_id WHERE p.name = '{OPTIMIZER_PARAM}' AND c.name IN ('mz_catalog_server', 'quickstart') ORDER BY c.name",
+                        "mz_catalog_server true",
+                        "quickstart false",
+                        f"> SELECT cr.name, p.value FROM mz_internal.mz_replica_system_parameters p JOIN mz_cluster_replicas cr ON cr.id = p.replica_id JOIN mz_clusters c ON c.id = cr.cluster_id WHERE p.name = '{LGALLOC_PARAM}' AND c.name IN ('ld_legacy', 'quickstart') ORDER BY c.name",
+                        "r1 false",
+                        "r1 true",
+                    ]
+                )
+            )
+            c.stop("materialized")
+    finally:
+        for flag in (LD_OPTIMIZER_FLAG_KEY, LD_LGALLOC_FLAG_KEY):
+            try:
+                ld_client.delete_flag(flag)
+            except:
+                pass  # ignore exceptions on cleanup
+
+
 class LaunchDarklyClient:
     """
     A test-specific LaunchDarkly client that simulates a client modifying
@@ -258,7 +433,21 @@ def __init__(
         self.project_key = project_key
         self.environment_key = environment_key
 
-    def create_flag(self, feature_flag_key: str, tags: list[str] = []) -> Any:
+    def create_flag(
+        self,
+        feature_flag_key: str,
+        tags: list[str] = [],
+        variations: list[Any] | None = None,
+        off_variation: int = 0,
+        on_variation: int = 1,
+    ) -> Any:
+        if variations is None:
+            variations = [
+                Variation(value=1073741824, name="1 GiB"),
+                Variation(value=2147483648, name="2 GiB"),
+                Variation(value=3221225472, name="3 GiB"),
+                Variation(value=4294967295, name="4 GiB - 1 (max size)"),
+            ]
         with launchdarkly_api.ApiClient(self.configuration) as api_client:
             api = feature_flags_api.FeatureFlagsApi(api_client)
             return api.post_feature_flag(
@@ -270,17 +459,12 @@ def create_flag(self, feature_flag_key: str, tags: list[str] = []) -> Any:
                         using_environment_id=True,
                         using_mobile_key=True,
                     ),
-                    variations=[
-                        Variation(value=1073741824, name="1 GiB"),
-                        Variation(value=2147483648, name="2 GiB"),
-                        Variation(value=3221225472, name="3 GiB"),
-                        Variation(value=4294967295, name="4 GiB - 1 (max size)"),
-                    ],
+                    variations=variations,
                     temporary=False,
                     tags=tags,
                     defaults=Defaults(
-                        off_variation=0,
-                        on_variation=1,
+                        off_variation=off_variation,
+                        on_variation=on_variation,
                     ),
                 ),
             )
@@ -290,6 +474,7 @@ def update_targeting(
         feature_flag_key: str,
         on: bool | None = None,
         contextTargets: list[Any] | None = None,
+        rules: list[Any] | None = None,
     ) -> Any:
         with launchdarkly_api.ApiClient(self.configuration) as api_client:
             api = feature_flags_api.FeatureFlagsApi(api_client)
@@ -322,6 +507,17 @@ def update_targeting(
                                     if contextTargets is not None
                                     else []
                                 ),
+                                (
+                                    [
+                                        PatchOperation(
+                                            op="replace",
+                                            path=f"/environments/{self.environment_key}/rules",
+                                            value=rules,
+                                        ),
+                                    ]
+                                    if rules is not None
+                                    else []
+                                ),
                             )
                         )
                     )