Experimental: the fermion arbiter, a commit sequencer for photon rooms

Behind experimental_features.fermion_arbiter_enabled (advertised as
org.matrix.experimental.fermion.arbiter). Members of a photon room report
each MLS commit they make as {arbiter, base, epoch, event_id,
epoch_interval_ms} to their own server at
POST /_matrix/client/unstable/org.matrix.experimental.fermion.v1/rooms/{roomId}/commits,
which answers itself if it is the room's arbiter and otherwise relays over
federation (PUT .../unstable/org.matrix.experimental.fermion.v1/rooms/{roomId}/commits).
The first commit seen on the current head is canonical; the arbiter signs
rulings naming the canonical extremity and the commits that reached it. At
epoch_interval_ms 0 the ruling is the response (a sequencer: vanilla MLS);
above it rulings are minted that long after the first unruled commit and
fetched from GET .../rulings?arbiter=&from=. The arbiter never joins the
group or sees a commit's bytes.

rust/src/config/mod.rs

@@ -77,4 +77,5 @@ pub struct ExperimentalConfig {
     pub msc4446_enabled: bool,
     pub msc4502_enabled: bool,
     pub photon_shared_to_device_enabled: bool,
+    pub fermion_arbiter_enabled: bool,
 }

rust/src/handlers/versions.rs

@@ -285,6 +285,9 @@ pub struct UnstableFeatureMap {
     /// Photon: shared to-device content with cached recipient sets
     #[serde(rename = "org.matrix.experimental.photon.shared_to_device")]
     photon_shared_to_device: bool,
+    /// Fermion: this server arbitrates the photon rooms its users create
+    #[serde(rename = "org.matrix.experimental.fermion.arbiter")]
+    fermion_arbiter: bool,
 
     // Whether new rooms will be set to encrypted or not (based on presets).
     #[serde(rename = "io.element.e2ee_forced.public")]
@@ -340,6 +343,7 @@ pub fn synapse_config_to_global_unstable_feature_map(
         msc4446_enabled: config.experimental.msc4446_enabled,
         msc4502: config.experimental.msc4502_enabled,
         photon_shared_to_device: config.experimental.photon_shared_to_device_enabled,
+        fermion_arbiter: config.experimental.fermion_arbiter_enabled,
         e2ee_forced_public: config
             .room
             .encryption_enabled_by_default_for_room_presets

synapse/config/experimental.py

@@ -324,3 +324,9 @@ class ExperimentalConfig(Config):
         self.photon_shared_to_device_enabled: bool = experimental.get(
             "photon_shared_to_device_enabled", False
         )
+
+        # Fermion: this server arbitrates (orders the MLS commits of) the
+        # photon rooms its users create.
+        self.fermion_arbiter_enabled: bool = experimental.get(
+            "fermion_arbiter_enabled", False
+        )

synapse/federation/transport/server/__init__.py

@@ -342,5 +342,21 @@ def register_servlets(
                 server_name=hs.hostname,
             ).register(resource)
 
+    if (
+        "federation" in servlet_groups
+        and hs.config.experimental.fermion_arbiter_enabled
+    ):
+        from synapse.federation.transport.server.fermion import (
+            FERMION_SERVLET_CLASSES,
+        )
+
+        for servletclass in FERMION_SERVLET_CLASSES:
+            servletclass(
+                hs=hs,
+                authenticator=authenticator,
+                ratelimiter=ratelimiter,
+                server_name=hs.hostname,
+            ).register(resource)
+
     if "federation" in servlet_groups:
         appservice_proxy.register_servlets(hs, resource, authenticator, ratelimiter)

synapse/federation/transport/server/fermion.py

@@ -0,0 +1,88 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2026 New Vector, Ltd
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# See the GNU Affero General Public License for more details:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+#
+
+"""The fermion arbiter over federation: a member's homeserver relays its
+submissions here (see `synapse.rest.client.fermion`)."""
+
+from typing import TYPE_CHECKING, Literal
+
+from synapse.api.urls import FEDERATION_UNSTABLE_PREFIX
+from synapse.federation.transport.server._base import (
+    Authenticator,
+    BaseFederationServlet,
+)
+from synapse.http.servlet import parse_integer_from_args
+from synapse.types import JsonDict
+
+if TYPE_CHECKING:
+    from synapse.server import HomeServer
+
+PATH = "/org.matrix.experimental.fermion.v1/rooms/(?P<room_id>[^/]*)"
+
+
+class FermionCommitsServlet(BaseFederationServlet):
+    PREFIX = FEDERATION_UNSTABLE_PREFIX
+    PATH = PATH + "/commits"
+    CATEGORY = "Federation requests"
+
+    def __init__(
+        self,
+        hs: "HomeServer",
+        authenticator: Authenticator,
+        ratelimiter: object,
+        server_name: str,
+    ):
+        super().__init__(hs, authenticator, ratelimiter, server_name)  # type: ignore[arg-type]
+        self.arbiter = hs.get_fermion_arbiter()
+
+    async def on_PUT(
+        self,
+        origin: str,
+        content: JsonDict,
+        query: dict[bytes, list[bytes]],
+        room_id: str,
+    ) -> tuple[int, JsonDict]:
+        return 200, self.arbiter.submit(room_id, content, origin)
+
+
+class FermionRulingsServlet(BaseFederationServlet):
+    PREFIX = FEDERATION_UNSTABLE_PREFIX
+    PATH = PATH + "/rulings"
+    CATEGORY = "Federation requests"
+
+    def __init__(
+        self,
+        hs: "HomeServer",
+        authenticator: Authenticator,
+        ratelimiter: object,
+        server_name: str,
+    ):
+        super().__init__(hs, authenticator, ratelimiter, server_name)  # type: ignore[arg-type]
+        self.arbiter = hs.get_fermion_arbiter()
+
+    async def on_GET(
+        self,
+        origin: str,
+        content: Literal[None],
+        query: dict[bytes, list[bytes]],
+        room_id: str,
+    ) -> tuple[int, JsonDict]:
+        since = parse_integer_from_args(query, "from", default=0)
+        return 200, {"rulings": self.arbiter.rulings(room_id, since)}
+
+
+FERMION_SERVLET_CLASSES: tuple[type[BaseFederationServlet], ...] = (
+    FermionCommitsServlet,
+    FermionRulingsServlet,
+)

synapse/rest/__init__.py

@@ -64,6 +64,7 @@ from synapse.rest.client import (
     room_keys,
     room_membership,
     room_upgrade_rest_servlet,
+    fermion,
     sendtodevice,
     sync,
     tags,
@@ -117,6 +118,7 @@ CLIENT_SERVLET_FUNCTIONS: tuple[RegisterServletsFunc, ...] = (
     devices.register_servlets,
     thirdparty.register_servlets,
     sendtodevice.register_servlets,
+    fermion.register_servlets,
     user_directory.register_servlets,
     room_upgrade_rest_servlet.register_servlets,
     capabilities.register_servlets,

synapse/rest/client/fermion.py

@@ -0,0 +1,235 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2026 New Vector, Ltd
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# See the GNU Affero General Public License for more details:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+#
+
+"""The fermion arbiter: ERA's finality node for a photon room created on
+this server (`docs/boson/PLAN.md` in matrix-rust-sdk).
+
+The server never joins the room's MLS group and is never sent a commit; it
+sees only epoch ids and event ids. Members report each commit they make
+(`base`, the DMLS epoch it is built on, and `epoch`, the one it starts),
+the server keeps the first commit it saw on its current head as canonical,
+and it signs rulings naming the canonical extremity and the commits that
+reached it. At `epoch_interval_ms` 0 the ruling comes back in the
+response, so a committer can ask before fanning out and drop a losing
+commit (vanilla MLS with a sequencer); above 0 rulings are minted that
+long after the first unruled commit, and members merge their losing
+branches onto the canonical one.
+
+A member talks to its own homeserver (the client endpoints here, with its
+access token), which answers itself if it is the room's arbiter and
+otherwise relays over federation (the servlets in
+`synapse.federation.transport.server.fermion`); so the arbiter knows the
+server and user behind every submission, and a browser never needs to
+reach a foreign server.
+
+ponytail: in-memory, per room, bounded by an LRU; a restart forgets the
+head and the next submission re-seeds it.
+"""
+
+import logging
+from typing import TYPE_CHECKING
+
+from signedjson.sign import sign_json
+
+from synapse.api.errors import SynapseError
+from synapse.api.urls import FEDERATION_UNSTABLE_PREFIX
+from synapse.http import servlet
+from synapse.http.server import HttpServer
+from synapse.http.servlet import (
+    assert_params_in_dict,
+    parse_integer,
+    parse_json_object_from_request,
+    parse_string,
+)
+from synapse.http.site import SynapseRequest
+from synapse.types import JsonDict
+from synapse.util.caches.lrucache import LruCache
+from synapse.util.duration import Duration
+
+from ._base import client_patterns
+
+if TYPE_CHECKING:
+    from synapse.server import HomeServer
+
+logger = logging.getLogger(__name__)
+
+PREFIX = "/org.matrix.experimental.fermion.v1/rooms/(?P<room_id>[^/]*)"
+FEDERATION_PATH = (
+    FEDERATION_UNSTABLE_PREFIX + "/org.matrix.experimental.fermion.v1/rooms/"
+)
+
+# Rulings kept per room for `GET …/rulings?from=`.
+RULINGS_KEPT = 64
+MAX_ROOMS = 10_000
+
+
+class ArbiterRoom:
+    """What the arbiter remembers of one room."""
+
+    def __init__(self) -> None:
+        # The canonical extremity: the DMLS epoch id commits must build on.
+        # None until the first submission seeds it with its base.
+        self.head: str | None = None
+        # Event ids of the commits accepted since the last ruling.
+        self.accepted: list[str] = []
+        self.rulings: list[JsonDict] = []
+        # When the first unruled commit arrived, if a ruling is due.
+        self.pending_since: int | None = None
+
+
+class Arbiter:
+    def __init__(self, hs: "HomeServer"):
+        self.clock = hs.get_clock()
+        self.server_name = hs.hostname
+        self.signing_key = hs.signing_key
+        self.rooms: LruCache[str, ArbiterRoom] = LruCache(
+            max_size=MAX_ROOMS,
+            clock=self.clock,
+            server_name=self.server_name,
+            cache_name="fermion_arbiter_rooms",
+        )
+
+    def room(self, room_id: str) -> ArbiterRoom:
+        room = self.rooms.get(room_id)
+        if room is None:
+            room = ArbiterRoom()
+            self.rooms[room_id] = room
+        return room
+
+    def submit(self, room_id: str, body: JsonDict, origin: str) -> JsonDict:
+        """Record a commit reported by `origin`; whether it is canonical,
+        and the ruling if one is minted now."""
+        assert_params_in_dict(body, ["base", "epoch", "event_id"])
+        base, epoch, event_id = body["base"], body["epoch"], body["event_id"]
+        if not all(isinstance(v, str) for v in (base, epoch, event_id)):
+            raise SynapseError(400, "'base', 'epoch' and 'event_id' must be strings")
+        interval = body.get("epoch_interval_ms", 0)
+        if not isinstance(interval, int) or interval < 0:
+            raise SynapseError(
+                400, "'epoch_interval_ms' must be a non-negative integer"
+            )
+        room = self.room(room_id)
+        if room.head is None:
+            room.head = base
+        accepted = base == room.head
+        if accepted:
+            room.head = epoch
+            room.accepted.append(event_id)
+        logger.info(
+            "fermion arbiter: %s commit %s on %s in %s from %s",
+            "accepted" if accepted else "rejected",
+            event_id,
+            base,
+            room_id,
+            origin,
+        )
+        if interval == 0:
+            if accepted or not room.rulings:
+                self.rule(room_id, room)
+            return {"accepted": accepted, "ruling": room.rulings[-1]}
+        if accepted and room.pending_since is None:
+            room.pending_since = self.clock.time_msec()
+            self.clock.call_later(
+                Duration(milliseconds=interval), self.rule, room_id, room
+            )
+        return {"accepted": accepted, "ruling": None}
+
+    def rule(self, room_id: str, room: ArbiterRoom) -> None:
+        """Mint a ruling: the canonical extremity and the commits accepted
+        since the last one, signed by this server."""
+        ruling = {
+            "room_id": room_id,
+            "index": len(room.rulings),
+            "canonical": room.head,
+            "commits": room.accepted,
+            "ts": self.clock.time_msec(),
+        }
+        room.rulings.append(sign_json(ruling, self.server_name, self.signing_key))
+        del room.rulings[:-RULINGS_KEPT]
+        room.accepted = []
+        room.pending_since = None
+
+    def rulings(self, room_id: str, since: int) -> list[JsonDict]:
+        room = self.rooms.get(room_id)
+        if room is None:
+            return []
+        return [r for r in room.rulings if r["index"] >= since]
+
+
+class ArbiterCommitsServlet(servlet.RestServlet):
+    """`POST …/rooms/{roomId}/commits` with `{arbiter, base, epoch, event_id,
+    epoch_interval_ms}`: report a commit to the room's arbiter server."""
+
+    PATTERNS = client_patterns(PREFIX + "/commits$", releases=(), unstable=True)
+    CATEGORY = "Encryption requests"
+
+    def __init__(self, hs: "HomeServer", arbiter: Arbiter):
+        super().__init__()
+        self.hs = hs
+        self.auth = hs.get_auth()
+        self.arbiter = arbiter
+        self.federation = hs.get_federation_http_client()
+
+    async def on_POST(
+        self, request: SynapseRequest, room_id: str
+    ) -> tuple[int, JsonDict]:
+        requester = await self.auth.get_user_by_req(request)
+        body = parse_json_object_from_request(request)
+        assert_params_in_dict(body, ["arbiter"])
+        destination = body.pop("arbiter")
+        if not isinstance(destination, str):
+            raise SynapseError(400, "'arbiter' must be a server name")
+        body["sender"] = requester.user.to_string()
+        if destination == self.hs.hostname:
+            return 200, self.arbiter.submit(room_id, body, self.hs.hostname)
+        return 200, await self.federation.put_json(
+            destination, FEDERATION_PATH + room_id + "/commits", data=body
+        )
+
+
+class ArbiterRulingsServlet(servlet.RestServlet):
+    """`GET …/rooms/{roomId}/rulings?arbiter=&from=`: the room's rulings
+    since index `from`."""
+
+    PATTERNS = client_patterns(PREFIX + "/rulings$", releases=(), unstable=True)
+    CATEGORY = "Encryption requests"
+
+    def __init__(self, hs: "HomeServer", arbiter: Arbiter):
+        super().__init__()
+        self.hs = hs
+        self.auth = hs.get_auth()
+        self.arbiter = arbiter
+        self.federation = hs.get_federation_http_client()
+
+    async def on_GET(
+        self, request: SynapseRequest, room_id: str
+    ) -> tuple[int, JsonDict]:
+        await self.auth.get_user_by_req(request)
+        destination = parse_string(request, "arbiter", required=True)
+        since = parse_integer(request, "from", default=0)
+        if destination == self.hs.hostname:
+            return 200, {"rulings": self.arbiter.rulings(room_id, since)}
+        return 200, await self.federation.get_json(
+            destination,
+            FEDERATION_PATH + room_id + "/rulings",
+            args={"from": str(since)},
+        )
+
+
+def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None:
+    if not hs.config.experimental.fermion_arbiter_enabled:
+        return
+    arbiter = hs.get_fermion_arbiter()
+    ArbiterCommitsServlet(hs, arbiter).register(http_server)
+    ArbiterRulingsServlet(hs, arbiter).register(http_server)

synapse/server.py

@@ -198,6 +198,7 @@ if TYPE_CHECKING:
     from synapse.handlers.jwt import JwtHandler
     from synapse.handlers.oidc import OidcHandler
     from synapse.handlers.saml import SamlHandler
+    from synapse.rest.client.fermion import Arbiter as FermionArbiter
     from synapse.storage._base import SQLBaseStore
 
 
@@ -889,6 +890,12 @@ class HomeServer(metaclass=abc.ABCMeta):
 
         return DeviceHandler(self)
 
+    @cache_in_self
+    def get_fermion_arbiter(self) -> "FermionArbiter":
+        from synapse.rest.client.fermion import Arbiter
+
+        return Arbiter(self)
+
     @cache_in_self
     def get_device_message_handler(self) -> DeviceMessageHandler:
         return DeviceMessageHandler(self)

tests/rest/client/test_fermion.py

@@ -0,0 +1,171 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2026 New Vector, Ltd
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# See the GNU Affero General Public License for more details:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+#
+
+from unittest.mock import AsyncMock
+
+from signedjson.key import get_verify_key
+from signedjson.sign import verify_signed_json
+
+from twisted.test.proto_helpers import MemoryReactor
+
+from synapse.rest import admin
+from synapse.rest.client import fermion, login
+from synapse.server import HomeServer
+from synapse.types import JsonDict
+from synapse.util.clock import Clock
+
+from tests.unittest import FederatingHomeserverTestCase
+
+ROOM = "/_matrix/client/unstable/org.matrix.experimental.fermion.v1/rooms/!r:test"
+FED_ROOM = (
+    "/_matrix/federation/unstable/org.matrix.experimental.fermion.v1/rooms/!r:test"
+)
+
+
+class FermionArbiterTestCase(FederatingHomeserverTestCase):
+    """The arbiter orders the commits members report and signs rulings;
+    a member's server relays to the room's arbiter server."""
+
+    servlets = [
+        admin.register_servlets,
+        login.register_servlets,
+        fermion.register_servlets,
+    ]
+
+    def default_config(self) -> JsonDict:
+        config = super().default_config()
+        config.setdefault("experimental_features", {"fermion_arbiter_enabled": True})
+        return config
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        super().prepare(reactor, clock, hs)
+        self.hs = hs
+        self.register_user("a", "pass")
+        self.tok = self.login("a", "pass")
+
+    def submit(
+        self,
+        base: str,
+        epoch: str,
+        event_id: str,
+        interval: int = 0,
+        arbiter: str = "test",
+    ) -> JsonDict:
+        channel = self.make_request(
+            "POST",
+            ROOM + "/commits",
+            content={
+                "arbiter": arbiter,
+                "base": base,
+                "epoch": epoch,
+                "event_id": event_id,
+                "epoch_interval_ms": interval,
+            },
+            access_token=self.tok,
+        )
+        self.assertEqual(channel.code, 200, channel.json_body)
+        return channel.json_body
+
+    def rulings(self, since: int = 0) -> list[JsonDict]:
+        channel = self.make_request(
+            "GET", ROOM + f"/rulings?arbiter=test&from={since}", access_token=self.tok
+        )
+        self.assertEqual(channel.code, 200, channel.json_body)
+        return channel.json_body["rulings"]
+
+    def test_sequencer_accepts_the_first_commit_on_the_head(self) -> None:
+        # The first submission seeds the head with its base.
+        first = self.submit("e0", "e1", "$c1")
+        self.assertTrue(first["accepted"])
+        self.assertEqual(first["ruling"]["canonical"], "e1")
+        self.assertEqual(first["ruling"]["commits"], ["$c1"])
+        self.assertEqual(first["ruling"]["index"], 0)
+        verify_signed_json(
+            first["ruling"], self.hs.hostname, get_verify_key(self.hs.signing_key)
+        )
+
+        # A concurrent commit on e0 loses, and learns the head from the ruling.
+        loser = self.submit("e0", "e1x", "$c1x")
+        self.assertFalse(loser["accepted"])
+        self.assertEqual(loser["ruling"]["canonical"], "e1")
+        self.assertEqual(loser["ruling"]["index"], 0)
+
+        # Building on the head wins again.
+        second = self.submit("e1", "e2", "$c2")
+        self.assertTrue(second["accepted"])
+        self.assertEqual(second["ruling"]["canonical"], "e2")
+        self.assertEqual(second["ruling"]["commits"], ["$c2"])
+        self.assertEqual([r["index"] for r in self.rulings()], [0, 1])
+        self.assertEqual([r["index"] for r in self.rulings(since=1)], [1])
+
+    def test_interval_batches_commits_into_one_ruling(self) -> None:
+        self.assertEqual(self.submit("e0", "e1", "$c1", interval=5000)["ruling"], None)
+        self.assertEqual(self.submit("e1", "e2", "$c2", interval=5000)["ruling"], None)
+        fork = self.submit("e1", "e2x", "$c2x", interval=5000)
+        self.assertFalse(fork["accepted"])
+        self.assertEqual(self.rulings(), [])
+        self.reactor.advance(5)
+        (ruling,) = self.rulings()
+        self.assertEqual(ruling["canonical"], "e2")
+        self.assertEqual(ruling["commits"], ["$c1", "$c2"])
+        # The loser rebases: a resolution on the head is canonical.
+        self.assertTrue(self.submit("e2", "e3", "$c3", interval=5000)["accepted"])
+        self.reactor.advance(5)
+        self.assertEqual(self.rulings(since=1)[0]["commits"], ["$c3"])
+
+    def test_over_federation(self) -> None:
+        # Another server's member reports to us, the arbiter.
+        channel = self.make_signed_federation_request(
+            "PUT",
+            FED_ROOM + "/commits",
+            content={
+                "base": "e0",
+                "epoch": "e1",
+                "event_id": "$c1",
+                "sender": "@b:other",
+            },
+        )
+        self.assertEqual(channel.code, 200, channel.json_body)
+        self.assertTrue(channel.json_body["accepted"])
+        channel = self.make_signed_federation_request(
+            "GET", FED_ROOM + "/rulings?from=0"
+        )
+        self.assertEqual(channel.code, 200, channel.json_body)
+        self.assertEqual(channel.json_body["rulings"][0]["commits"], ["$c1"])
+
+        # Our member reports to another server's arbiter: relayed, with the
+        # sender filled in.
+        client = self.hs.get_federation_http_client()
+        client.put_json = AsyncMock(return_value={"accepted": True, "ruling": None})  # type: ignore[method-assign]
+        self.assertTrue(self.submit("x0", "x1", "$d1", arbiter="other")["accepted"])
+        destination, path = client.put_json.call_args.args
+        self.assertEqual(destination, "other")
+        self.assertEqual(path, FED_ROOM + "/commits")
+        body = client.put_json.call_args.kwargs["data"]
+        self.assertEqual(body["sender"], "@a:test")
+        self.assertNotIn("arbiter", body)
+
+    def test_bad_submissions(self) -> None:
+        for content in (
+            {"base": "e0"},
+            {"arbiter": "test", "base": "e0"},
+            {"arbiter": "test", "base": "e0", "epoch": "e1", "event_id": 1},
+        ):
+            channel = self.make_request(
+                "POST", ROOM + "/commits", content=content, access_token=self.tok
+            )
+            self.assertEqual(channel.code, 400, content)
+        channel = self.make_request("POST", ROOM + "/commits", content={})
+        self.assertEqual(channel.code, 401)
+        self.assertEqual(self.rulings(), [])