mirror of
https://github.com/element-hq/synapse.git
synced 2026-08-05 15:20:14 +00:00
fix: provide remote servers a way to find out about an event created during the remote join handshake (#19390)
This commit is contained in:
FrenchGithubUser
@@ -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.
|
||||
@@ -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
|
||||
|
||||
@@ -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=
|
||||
|
||||
@@ -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(),
|
||||
)
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
Reference in New Issue
Block a user