Merge pull request #632 from element-hq/bbz/fix-routing-to-initial-sync

Synapse: fix requests being routed to initial-synchrotron incorrectly

Author: benbz

.github/workflows/pytest.yml

@@ -105,7 +105,7 @@ jobs:
 
     - name: On upgrade, Synapse can restart, expect 429 to occur
       run: |
-        export PYTEST_EXPECTED_HTTP_STATUS_CODES="429"
+        echo "PYTEST_EXPECTED_HTTP_STATUS_CODES=429" >> "$GITHUB_ENV"
       if: ${{ matrix.test-from-ref != github.event.pull_request.head.sha }}
 
     - name: Test with pytest (upgrade or idempotent setup)
@@ -136,7 +136,7 @@ jobs:
       if: ${{ failure() }}
       uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02  # v4
       with:
-        name: ess-helm-logs-${{ matrix.test-component }}
+        name: ess-helm-logs-${{ matrix.test-component }}-${{ matrix.test-from-ref }}
         path: ess-helm-logs
         retention-days: 1
 

charts/matrix-stack/ci/fragments/synapse-pytest-self-extras.yaml

@@ -20,10 +20,16 @@ synapse:
     # A non-HTTP worker & a stream writer
     event-persister:
       enabled: true
+    # initial-synchrotron & synchrotron have non-trivial routing behaviour
+    initial-synchrotron:
+      enabled: true
     # A standard HTTP worker
     sliding-sync:
       replicas: 2
       enabled: true
+    # initial-synchrotron & synchrotron have non-trivial routing behaviour
+    synchrotron:
+      enabled: true
     # Media repo is fairly distinct from other workers
     media-repository:
       enabled: true

charts/matrix-stack/ci/pytest-synapse-values.yaml

@@ -58,12 +58,18 @@ synapse:
     # A non-HTTP worker & a stream writer
     event-persister:
       enabled: true
+    # initial-synchrotron & synchrotron have non-trivial routing behaviour
+    initial-synchrotron:
+      enabled: true
     # Media repo is fairly distinct from other workers
     media-repository:
       enabled: true
     # A standard HTTP worker
     sliding-sync:
       enabled: true
       replicas: 2
+    # initial-synchrotron & synchrotron have non-trivial routing behaviour
+    synchrotron:
+      enabled: true
 wellKnownDelegation:
   enabled: false

charts/matrix-stack/configs/synapse/partial-haproxy.cfg.tpl

@@ -74,18 +74,18 @@ frontend synapse-http-in
   http-request set-var(req.backend) path,map_reg(/synapse/path_map_file,main) unless { var(req.backend) -m found }
 {{- if dig "initial-synchrotron" "enabled" false .workers }}
 
-  acl is_initial_sync path -m reg ^/_matrix/client/(api/v1|r0|v3)/initialSync$
-  acl is_initial_sync path -m reg ^/_matrix/client/(api/v1|r0|v3)/rooms/[^/]+/initialSync$
-  # https://spec.matrix.org/v1.14/client-server-api/#get_matrixclientv3sync
-  acl is_initial_sync path -m reg ^/_matrix/client/(r0|v3)/sync$ { urlp("full_state") -m str true }
-  acl is_initial_sync path -m reg ^/_matrix/client/(r0|v3)/sync$ !{ urlp("since") -m found }
-  # https://spec.matrix.org/latest/client-server-api/#get_matrixclientv3events
-  acl is_initial_sync path -m reg ^/_matrix/client/(api/v1|r0|v3)/events$ !{ urlp("from") -m found }
+  acl has_available_initial_syncs nbsrv('synapse-initial-synchrotron') ge 1
 
   # Set to the initial-synchrotron backend if it is one of these magic paths AND we have workers in the initial-synchrotron backend
   # This means that we don't update the backend from synchrotron if that's configured but there's no initial-synchrotron servers available
   # And then can it fallback to main if there are no synchrotron servers either
-  http-request set-var(req.backend) str('initial-synchrotron') if is_initial_sync { nbsrv('synapse-initial-synchrotron') ge 1 }
+  http-request set-var(req.backend) str('initial-synchrotron') if has_available_initial_syncs { path -m reg ^/_matrix/client/(api/v1|r0|v3)/initialSync$ }
+  http-request set-var(req.backend) str('initial-synchrotron') if has_available_initial_syncs { path -m reg ^/_matrix/client/(api/v1|r0|v3)/rooms/[^/]+/initialSync$ }
+  # https://spec.matrix.org/v1.14/client-server-api/#get_matrixclientv3sync
+  http-request set-var(req.backend) str('initial-synchrotron') if has_available_initial_syncs { path -m reg ^/_matrix/client/(r0|v3)/sync$ } { urlp("full_state") -m str true }
+  http-request set-var(req.backend) str('initial-synchrotron') if has_available_initial_syncs { path -m reg ^/_matrix/client/(r0|v3)/sync$ } !{ urlp("since") -m found }
+  # https://spec.matrix.org/latest/client-server-api/#get_matrixclientv3events
+  http-request set-var(req.backend) str('initial-synchrotron') if has_available_initial_syncs { path -m reg ^/_matrix/client/(api/v1|r0|v3)/events$ } !{ urlp("from") -m found }
 {{- end }}
 
 {{- if include "element-io.matrix-authentication-service.readyToHandleAuth" (dict "root" $root) }}

newsfragments/632.fixed.md

@@ -0,0 +1 @@
+Synapse: fix requests being routed to initial-synchrotron incorrectly.

tests/integration/lib/utils.py

@@ -29,7 +29,6 @@
     }
     | set(int(code) for code in os.environ.get("PYTEST_EXPECTED_HTTP_STATUS_CODES", "").split(",") if code),
     retry_all_server_errors=False,
-    exceptions={aiohttp.client_exceptions.ClientResponseError},
 )
 
 

tests/integration/test_synapse.py

@@ -7,6 +7,9 @@
 
 import pyhelm3
 import pytest
+from aiohttp.client_exceptions import ClientResponseError
+from lightkube import AsyncClient
+from lightkube.resources.core_v1 import Pod
 
 from .fixtures import ESSData, User
 from .lib.helpers import deploy_with_values_patch, get_deployment_marker
@@ -35,7 +38,9 @@ async def test_synapse_can_access_client_api(
 @pytest.mark.skipif(value_file_has("synapse.enabled", False), reason="Synapse not deployed")
 @pytest.mark.parametrize("users", [(User(name="sliding-sync-user"),)], indirect=True)
 @pytest.mark.asyncio_cooperative
-async def test_simplified_sliding_sync_syncs(ssl_context, users, generated_data: ESSData):
+async def test_simplified_sliding_sync_syncs(ingress_ready, ssl_context, users, generated_data: ESSData):
+    await ingress_ready("synapse")
+
     access_token = users[0].access_token
 
     sync_result = await aiohttp_post_json(
@@ -48,6 +53,73 @@ async def test_simplified_sliding_sync_syncs(ssl_context, users, generated_data:
     assert "pos" in sync_result
 
 
+@pytest.mark.skipif(value_file_has("synapse.enabled", False), reason="Synapse not deployed")
+@pytest.mark.asyncio_cooperative
+async def test_routes_to_synapse_workers_correctly(
+    ingress_ready, kube_client: AsyncClient, ssl_context, generated_data: ESSData
+):
+    await ingress_ready("synapse")
+
+    main_backend = "main/main"
+    if value_file_has("synapse.workers.sliding-sync.enabled", True):
+        sliding_sync_backend = "sliding-sync/sliding-sync"
+    else:
+        sliding_sync_backend = main_backend
+
+    if value_file_has("synapse.workers.synchrotron.enabled", True):
+        synchrotron_backend = "synchrotron/synchrotron"
+    else:
+        synchrotron_backend = main_backend
+
+    if value_file_has("synapse.workers.initial-synchrotron.enabled", True):
+        initial_synchrotron_backend = "initial-synchrotron/initial-sync"
+    else:
+        initial_synchrotron_backend = synchrotron_backend
+
+    # We don't care about any of these succeeding, only that the requests are made and HAProxy dispatches correctly
+    # So no auth required and parameters can be made up
+    paths_to_backends = {
+        # initial-synchrotron
+        "/_matrix/client/v3/initialSync": initial_synchrotron_backend,
+        "/_matrix/client/v3/rooms/aroomid/initialSync": initial_synchrotron_backend,
+        "/_matrix/client/v3/sync?full_state=true": initial_synchrotron_backend,
+        "/_matrix/client/v3/sync": initial_synchrotron_backend,
+        "/_matrix/client/v3/events": initial_synchrotron_backend,
+        # synchrotron
+        "/_matrix/client/v3/sync?since=recently": synchrotron_backend,
+        "/_matrix/client/v3/events?from=recently": synchrotron_backend,
+        # sliding-sync
+        "/_matrix/client/unstable/org.matrix.simplified_msc3575/sync": sliding_sync_backend,
+        # Would be client-reader but not configured in tests
+        "/_matrix/client/versions": main_backend,
+    }
+
+    for path in paths_to_backends:
+        try:
+            await aiohttp_get_json(f"https://synapse.{generated_data.server_name}{path}", ssl_context)
+        except ClientResponseError as e:
+            # We can't use pytest.raises as no exception (200) is valid
+            assert e.status in [401, 405], f"{path} had an unexpected status. {e=}"  # noqa P1017
+
+    http_log_lines = []
+    async for haproxy_pod in kube_client.list(
+        Pod, namespace=generated_data.ess_namespace, labels={"app.kubernetes.io/name": "haproxy"}
+    ):
+        assert haproxy_pod.metadata
+        assert haproxy_pod.metadata.name
+        assert haproxy_pod.metadata.namespace
+        async for log_line in kube_client.log(haproxy_pod.metadata.name, namespace=haproxy_pod.metadata.namespace):
+            if "HTTP/1.1" in log_line:
+                http_log_lines.append(log_line)
+
+    for path, backend in paths_to_backends.items():
+        matching_lines = [line for line in http_log_lines if f"GET {path} HTTP/1.1" in line]
+
+        assert len(matching_lines) > 0, f"Requests for {path} did not appear in the HAProxy logs"
+        for matching_line in matching_lines:
+            assert f"synapse-http-in synapse-{backend}" in matching_line, f"{path} was routed unexpectedly"
+
+
 @pytest.mark.skipif(value_file_has("synapse.enabled", False), reason="Synapse not deployed")
 @pytest.mark.parametrize("users", [(User(name="media-upload-unauth"),)], indirect=True)
 @pytest.mark.asyncio_cooperative