diff --git a/changelog.d/19390.bugfix b/changelog.d/19390.bugfix new file mode 100644 index 0000000000..706f779174 --- /dev/null +++ b/changelog.d/19390.bugfix @@ -0,0 +1 @@ +Provide remote servers a way to find out about an event created during the remote join handshake. Contributed by @FrenchGithubUser and @jason-famedly @ Famedly. diff --git a/complement/go.mod b/complement/go.mod index 5ebf19bbf6..dfd8e73f22 100644 --- a/complement/go.mod +++ b/complement/go.mod @@ -31,6 +31,7 @@ require ( github.com/go-logr/logr v1.4.3 // indirect github.com/go-logr/stdr v1.2.2 // indirect github.com/gogo/protobuf v1.3.2 // indirect + github.com/gorilla/mux v1.8.0 // indirect github.com/matrix-org/gomatrix v0.0.0-20220926102614-ceba4d9f7530 // indirect github.com/matrix-org/util v0.0.0-20221111132719-399730281e66 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect diff --git a/complement/go.sum b/complement/go.sum index 25c6922669..325ad216c6 100644 --- a/complement/go.sum +++ b/complement/go.sum @@ -38,6 +38,8 @@ github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/gorilla/mux v1.8.0 h1:i40aqfkR1h2SlN9hojwV5ZA91wcXFOvkdNIeFDP5koI= +github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So= github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 h1:HWRh5R2+9EifMyIHV7ZV+MIZqgz+PMpZ14Jynv3O2Zs= github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0/go.mod h1:JfhWUomR1baixubs02l85lZYYOm7LV6om4ceouMv45c= github.com/h2non/parth v0.0.0-20190131123155-b4df798d6542 h1:2VTzZjLZBgl62/EtslCrtky5vbi9dd7HrQPQIx6wqiw= diff --git a/complement/tests/federation_room_join_test.go b/complement/tests/federation_room_join_test.go new file mode 100644 index 0000000000..7ed1f93f87 --- /dev/null +++ b/complement/tests/federation_room_join_test.go @@ -0,0 +1,178 @@ +package synapse_tests + +import ( + "context" + "io" + "net/http" + "sync/atomic" + "testing" + "time" + + "github.com/matrix-org/complement" + + "github.com/matrix-org/gomatrixserverlib" + + "github.com/tidwall/gjson" + + "github.com/matrix-org/complement/b" + "github.com/matrix-org/complement/federation" + "github.com/matrix-org/complement/helpers" + "github.com/matrix-org/complement/must" +) + +// This test verifies that events sent into a room between a /make_join and +// /send_join are not lost to the joining server. When an event is created +// during the join handshake, the join event's prev_events (set at make_join +// time) won't reference it, creating two forward extremities. The server +// handling the join should ensure the joining server can discover the missed +// event, for example by sending a follow-up event that references both +// extremities, prompting the joining server to backfill. +// +// See https://github.com/element-hq/synapse/pull/19390 +// +// This test lives as a in-repo Synapse Complement test because the spec doesn't mandate +// which events should be resolvable after the `/make_join`/`/send_join` dance (or that +// a homeserver should send `m.dummy` events to tie things together) +func TestEventBetweenMakeJoinAndSendJoinIsNotLost(t *testing.T) { + deployment := complement.Deploy(t, 1) + defer deployment.Destroy(t) + + alice := deployment.Register(t, "hs1", helpers.RegistrationOpts{}) + + // We track the message event ID sent between make_join and send_join. + // After send_join, we wait for hs1 to send us either: + // - the message event itself, or + // - any event whose prev_events reference the message (e.g. a dummy event) + // + // atomic.Value is used because messageEventID is written on the main goroutine and + // read on the HTTP handler goroutine, and needs synchronization (without + // synchronization, writes are not guaranteed to be observed by other goroutines) + var messageEventID atomic.Value + messageEventID.Store("") + messageDiscoverableWaiter := helpers.NewWaiter() + + srv := federation.NewServer(t, deployment, + // hs1 fetches our signing keys via /_matrix/key/v2/server to verify our + // identity before accepting federation requests. Without this handler, + // make_join is rejected with 401 M_UNAUTHORIZED. + federation.HandleKeyRequests(), + ) + // After send_join, hs1 will start sending us federation transactions via + // /_matrix/federation/v1/send/{txnID}. Since we handle /send manually + // below, any other requests (e.g. key fetches) that arrive unexpectedly + // should be tolerated rather than treated as test failures. + srv.UnexpectedRequestsAreErrors = false + + // Custom /send handler: hs1 will push new room events to us via federation + // transactions once we've joined. We use a raw handler because the + // Complement server is not fully in the room until send_join completes, so + // we can't use HandleTransactionRequests (which requires the room in + // srv.rooms). Instead we parse the raw transaction body ourselves. + srv.Mux().Handle("/_matrix/federation/v1/send/{transactionID}", http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + body, err := io.ReadAll(req.Body) + must.NotError(t, "failed to read request body in /send handler: %v", err) + txn := gjson.ParseBytes(body) + txn.Get("pdus").ForEach(func(_, pdu gjson.Result) bool { + eventID := pdu.Get("event_id").String() + eventType := pdu.Get("type").String() + t.Logf("Received PDU via /send: type=%s id=%s", eventType, eventID) + + // messageEventID is set after make_join but before send_join. + // Transactions can arrive before that window, so skip PDUs that + // arrive before we know which event to look for. + msgID, _ := messageEventID.Load().(string) + if msgID == "" { + return true + } + + // Check if this IS the message event (server pushed it directly). + if eventID == msgID { + messageDiscoverableWaiter.Finish() + return true + } + + // Check if this event's prev_events directly reference the message (e.g. a dummy + // event tying the two forward extremities together). If so, the joining server + // can backfill from that event and will discover the message. + // + // XXX: We only check one level of prev_events: if the reference is deeper in the + // DAG, it's valid and the joining server can still reach the message through + // backfill but our checks don't account for that yet (feel free to edit this + // assertion if you run into this) + pdu.Get("prev_events").ForEach(func(_, prevEvent gjson.Result) bool { + if prevEvent.String() == msgID { + messageDiscoverableWaiter.Finish() + return false + } + return true + }) + + return true + }) + w.WriteHeader(200) + // Respond with an empty PDU error map, which is the federation /send + // success response format: each key would be a PDU ID whose processing + // failed; an empty object means all PDUs were accepted. + w.Write([]byte(`{"pdus":{}}`)) + })).Methods("PUT") + + cancel := srv.Listen() + defer cancel() + + // Alice creates a room on hs1. + roomID := alice.MustCreateRoom(t, map[string]interface{}{ + "preset": "public_chat", + }) + + charlie := srv.UserID("charlie") + origin := srv.ServerName() + fedClient := srv.FederationClient(deployment) + + // Step 1: make_join, hs1 returns a join event template whose prev_events + // reflect the current room DAG tips. + makeJoinResp, err := fedClient.MakeJoin( + context.Background(), origin, + deployment.GetFullyQualifiedHomeserverName(t, "hs1"), + roomID, charlie, + ) + must.NotError(t, "MakeJoin", err) + + // Step 2: Alice sends a message on hs1. This advances the DAG past the + // point captured by make_join's prev_events. The Complement server is not + // yet in the room, so it won't receive this event via normal federation. + messageEventID.Store(alice.SendEventSynced(t, roomID, b.Event{ + Type: "m.room.message", + Content: map[string]interface{}{ + "msgtype": "m.text", + "body": "Message sent between make_join and send_join", + }, + })) + t.Logf("Alice sent message %s between make_join and send_join", messageEventID.Load()) + + // Step 3: Build and sign the join event, then send_join. + // The join event's prev_events are from step 1 (before the message), + // so persisting it on hs1 creates two forward extremities: the message + // and the join. + verImpl, err := gomatrixserverlib.GetRoomVersion(makeJoinResp.RoomVersion) + must.NotError(t, "GetRoomVersion", err) + eb := verImpl.NewEventBuilderFromProtoEvent(&makeJoinResp.JoinEvent) + joinEvent, err := eb.Build(time.Now(), srv.ServerName(), srv.KeyID, srv.Priv) + must.NotError(t, "Build join event", err) + + _, err = fedClient.SendJoin( + context.Background(), origin, + deployment.GetFullyQualifiedHomeserverName(t, "hs1"), + joinEvent, + ) + must.NotError(t, "SendJoin", err) + + // Step 4: hs1 should make the missed message discoverable to the joining + // server. We accept either receiving the message event directly, or + // receiving any event whose prev_events reference it (allowing the + // joining server to backfill). + messageDiscoverableWaiter.Waitf(t, 5*time.Second, + "Timed out waiting for message event %s to become discoverable — "+ + "the event sent between make_join and send_join was lost to the "+ + "joining server", messageEventID.Load(), + ) +} diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py index 6069287975..2fc78c012e 100644 --- a/synapse/federation/federation_server.py +++ b/synapse/federation/federation_server.py @@ -816,6 +816,15 @@ class FederationServer(FederationBase): event, context = await self._on_send_membership_event( origin, content, Membership.JOIN, room_id ) + # Use the join event's own stream ordering as the upper bound when fetching + # forward extremities (below), so we only consider extremities that existed at + # or before the join rather than those introduced by concurrent writes that + # occur while we prepare the response. + # Note: in workers mode the event is persisted on a separate worker, so + # event.internal_metadata.stream_ordering is not populated here; query the DB. + stream_ordering_of_join = ( + await self.store.get_position_for_event(event.event_id) + ).stream prev_state_ids = await context.get_prev_state_ids() @@ -856,6 +865,30 @@ class FederationServer(FederationBase): "members_omitted": caller_supports_partial_state, } + # Check the forward extremities for the room here. If there is more than one, it + # is likely that another event was created in the room during the + # make_join/send_join handshake. The joining server is likely to thus miss this event + # until a second event is created that references it - which could be some time. + # In that case, we proactively send a dummy extensible event that ties these + # forward extremities together. The remote server will then attempt to backfill + # the missing event on its own. + # + # By not sending the 'missing event' directly, but instead having the joining + # homeserver backfill it, the stream ordering for the missing event will be + # "before" the join (which is what we expect). + + forward_extremities = ( + await self.store.get_forward_extremities_for_room_at_stream_ordering( + room_id, stream_ordering_of_join + ) + ) + + if len(forward_extremities) > 1: + # The likelihood of this being used is extremely low, thus only build the handler + # when necessary. + _creation_handler = self.hs.get_event_creation_handler() + await _creation_handler._send_dummy_event_after_room_join(room_id) + if servers_in_room is not None: resp["servers_in_room"] = list(servers_in_room) diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py index ed04547d42..6bdef7e202 100644 --- a/synapse/handlers/message.py +++ b/synapse/handlers/message.py @@ -2295,7 +2295,32 @@ class EventCreationHandler: now = self.clock.time_msec() self._rooms_to_exclude_from_dummy_event_insertion[room_id] = now - async def _send_dummy_event_for_room(self, room_id: str) -> bool: + async def _send_dummy_event_after_room_join(self, room_id: str) -> None: + """ + Creates and sends a dummy event into the given room, referencing the + current forward extremities (via `prev_events`). + This should only be triggered when handling a remote join while events + were sent during the make_join/send_join handshake. The joining + homeserver would otherwise not immediately know to backfill those events + and would "miss" them. + """ + async with self._worker_lock_handler.acquire_read_write_lock( + NEW_EVENT_DURING_PURGE_LOCK_NAME, room_id, write=False + ): + dummy_event_sent = await self._send_dummy_event_for_room( + room_id, proactively_send=True + ) + + if not dummy_event_sent: + logger.warning( + "Failed to send dummy event into room %s after remote join; " + "no local user with permission was found", + room_id, + ) + + async def _send_dummy_event_for_room( + self, room_id: str, proactively_send: bool = False + ) -> bool: """Attempt to send a dummy event for the given room. Args: @@ -2327,8 +2352,7 @@ class EventCreationHandler: }, ) context = await unpersisted_context.persist(event) - - event.internal_metadata.proactively_send = False + event.internal_metadata.proactively_send = proactively_send # Since this is a dummy-event it is OK if it is sent by a # shadow-banned user.