test: Probe initial flag snapshots before checking configuration sync

test/launchdarkly/mzcompose.py

@@ -11,14 +11,16 @@
 Test the LaunchDarkly integration, get configuration flags from LD.
 """
 
+import json
 from itertools import chain
 from os import environ
 from textwrap import dedent
-from time import sleep
+from time import monotonic, sleep
 from typing import Any
 from uuid import uuid1
 
 import launchdarkly_api  # type: ignore
+import requests
 from launchdarkly_api.api import feature_flags_api  # type: ignore
 from launchdarkly_api.model.client_side_availability_post import (  # type: ignore
     ClientSideAvailabilityPost,
@@ -96,7 +98,8 @@ def context_rule(
         },
         external_metadata_store=True,
     ),
-    Testdrive(no_reset=True, seed=1),
+    # Include streaming reconnect backoff in the live-update assertion window.
+    Testdrive(no_reset=True, seed=1, default_timeout="120s"),
 ]
 
 
@@ -143,10 +146,6 @@ def workflow_default(c: Composition) -> None:
             on=True,
         )
 
-        # 3 seconds should be enough to avoid race conditions between the update
-        # above and the query below.
-        sleep(3)
-
         # Assert that the value is as expected after the initial parameter sync.
         with c.override(
             Materialized(
@@ -622,7 +621,7 @@ def update_targeting(
     ) -> Any:
         with launchdarkly_api.ApiClient(self.configuration) as api_client:
             api = feature_flags_api.FeatureFlagsApi(api_client)
-            return api.patch_feature_flag(
+            flag = api.patch_feature_flag(
                 project_key=self.project_key,
                 feature_flag_key=feature_flag_key,
                 patch_with_comment=PatchWithComment(
@@ -667,6 +666,73 @@ def update_targeting(
                     )
                 ),
             )
+        self.wait_for_propagation(
+            feature_flag_key, flag.environments[self.environment_key].version
+        )
+        return flag
+
+    def wait_for_propagation(self, feature_flag_key: str, version: int) -> None:
+        """Probe fresh SDK stream snapshots until one contains the updated flag."""
+        assert LAUNCHDARKLY_SDK_KEY is not None
+        # A successful management API write does not imply SDK visibility. In
+        # particular, the boot-only assertion has no sync loop to repair a stale
+        # initial read. Probe the same initial snapshot endpoint that boot uses,
+        # not the polling endpoint or later patches on an already-open stream.
+        # These versions are environment-specific, not the flag's top-level
+        # management API version.
+        deadline = monotonic() + 120
+        observed: int | str | None = "no snapshot"
+        while monotonic() < deadline:
+            try:
+                with requests.get(
+                    "https://stream.launchdarkly.com/all",
+                    headers={"Authorization": LAUNCHDARKLY_SDK_KEY},
+                    stream=True,
+                    timeout=10,
+                ) as response:
+                    response.raise_for_status()
+                    response.encoding = "utf-8"
+                    event = ""
+                    data: list[str] = []
+                    for line in response.iter_lines(decode_unicode=True):
+                        if monotonic() >= deadline:
+                            break
+                        if line.startswith("event:"):
+                            event = line[len("event:") :].strip()
+                        elif line.startswith("data:"):
+                            data.append(line[len("data:") :].lstrip())
+                        elif not line and data:
+                            snapshot = json.loads("\n".join(data))
+                            if event != "put" or snapshot["path"] != "/":
+                                raise UIError(
+                                    "Expected an initial root put from SDK stream"
+                                )
+                            flag = snapshot["data"]["flags"].get(feature_flag_key)
+                            snapshot_version = (
+                                flag["version"] if flag is not None else None
+                            )
+                            observed = snapshot_version
+                            if (
+                                snapshot_version is not None
+                                and snapshot_version >= version
+                            ):
+                                print(
+                                    f"SDK snapshot contains {feature_flag_key} version "
+                                    f"{observed} (wanted {version})"
+                                )
+                                return
+                            break
+            except requests.HTTPError as e:
+                if e.response is None or e.response.status_code < 500:
+                    raise
+                observed = f"HTTP {e.response.status_code}"
+            except (requests.ConnectionError, requests.Timeout) as e:
+                observed = type(e).__name__
+            sleep(1)
+        raise UIError(
+            f"SDK propagation timed out for {feature_flag_key}: "
+            f"expected version >= {version}, last observed {observed}"
+        )
 
     def delete_flag(self, feature_flag_key: str) -> Any:
         with launchdarkly_api.ApiClient(self.configuration) as api_client: