From 99bf0d6c4432d63b11cc6c33f67c2be53627b93c Mon Sep 17 00:00:00 2001 From: Kegan Dougal <7190048+kegsay@users.noreply.github.com> Date: Wed, 23 Sep 2026 16:34:06 +0100 Subject: [PATCH] Add minimal /send_join MSC4242 Complement tests This is to serve as regression tests for: - rejoining a room not working because we skipped seen events during processing - rejected events (due to cascading) being accepted because we didn't persist rejection status in-memory alongside the state group --- tests/msc4242/main_test.go | 11 ++ tests/msc4242/msc4242_joins_test.go | 289 ++++++++++++++++++++++++++++ tests/msc4242/state_dag_test.go | 254 ++++++++++++++++++++++++ 3 files changed, 554 insertions(+) create mode 100644 tests/msc4242/main_test.go create mode 100644 tests/msc4242/msc4242_joins_test.go create mode 100644 tests/msc4242/state_dag_test.go diff --git a/tests/msc4242/main_test.go b/tests/msc4242/main_test.go new file mode 100644 index 000000000..4c0663812 --- /dev/null +++ b/tests/msc4242/main_test.go @@ -0,0 +1,11 @@ +package tests + +import ( + "testing" + + "github.com/matrix-org/complement" +) + +func TestMain(m *testing.M) { + complement.TestMain(m, "msc4242") +} diff --git a/tests/msc4242/msc4242_joins_test.go b/tests/msc4242/msc4242_joins_test.go new file mode 100644 index 000000000..1a081f058 --- /dev/null +++ b/tests/msc4242/msc4242_joins_test.go @@ -0,0 +1,289 @@ +package tests + +import ( + "encoding/json" + "fmt" + "testing" + "time" + + "github.com/matrix-org/complement" + "github.com/matrix-org/complement/client" + "github.com/matrix-org/complement/federation" + "github.com/matrix-org/complement/helpers" + "github.com/matrix-org/gomatrixserverlib" + "github.com/matrix-org/gomatrixserverlib/spec" +) + +// Test that a homeserver can join a public state DAG room over federation, leave it, then rejoin it. +// +// The rejoin is the interesting part: the homeserver already has most of the state DAG persisted +// from the first join, so the /send_join response is a mixture of events it has seen before and +// events which were added to the room while it was away. +func TestMSC4242JoinLeaveRejoinPublicRoom(t *testing.T) { + deployment := complement.Deploy(t, 1) + defer deployment.Destroy(t) + alice := deployment.Register(t, "hs1", helpers.RegistrationOpts{}) + + srv := federation.NewServer(t, deployment, + federation.HandleKeyRequests(), + // accept incoming presence transactions, membership events, etc + federation.HandleTransactionRequests(nil, nil), + // accept incoming /event requests + federation.HandleEventRequests(), + federation.HandleMakeSendJoinRequests(), + ) + // the homeserver makes /query/profile and /user/devices requests we don't care about + srv.UnexpectedRequestsAreErrors = false + cancel := srv.Listen() + defer cancel() + + bob := srv.UserID("bob") + // InitialRoomEvents sets a join rule of "public", so no invite is needed to (re)join. + room := srv.MustMakeRoom(t, roomVersion, + federation.InitialRoomEvents(roomVersion, bob), + federation.WithImpl(ServerRoomImplStateDAG(t)), + ) + + // Lengthen the state DAG before the join so the homeserver has to walk it rather than just + // consume the room creation events. + setDisplayName(t, srv, room, bob, "bob before the join", 3) + mustSetRoomName(t, srv, room, bob, "before the join") + + // Alice should be able to join this room. + alice.MustJoinRoom(t, room.RoomID, []spec.ServerName{srv.ServerName()}) + sinceJoined := alice.MustSyncUntil(t, client.SyncReq{}, client.SyncJoinedTo(alice.UserID, room.RoomID)) + mustHaveStateEventContent( + t, currentRoomState(t, alice, room.RoomID), spec.MRoomName, "", "name", "before the join", + "current state at first join missing room name", + ) + + // Alice leaves, wait for it to propagate. + alice.MustLeaveRoom(t, room.RoomID) + alice.MustSyncUntil(t, client.SyncReq{Since: sinceJoined}, client.SyncLeftFrom(alice.UserID, room.RoomID)) + leaveEvent := awaitMembership(t, room, alice.UserID, "leave") + t.Logf("state DAG: %s = (m.room.member, %s) leave prev_state_events=%v prev_events=%v", + leaveEvent.EventID(), alice.UserID, leaveEvent.PrevStateEventIDs(), leaveEvent.PrevEventIDs()) + + // Add more state while the homeserver is not in the room + setDisplayName(t, srv, room, bob, "bob after the leave", 3) + mustSetRoomName(t, srv, room, bob, "after the leave") + + // Alice rejoins the room + alice.MustJoinRoom(t, room.RoomID, []spec.ServerName{srv.ServerName()}) + alice.MustSyncUntil(t, client.SyncReq{}, client.SyncJoinedTo(alice.UserID, room.RoomID)) + rejoinEvent := awaitMembership(t, room, alice.UserID, "join") + t.Logf("state DAG: %s = (m.room.member, %s) rejoin prev_state_events=%v prev_events=%v", + rejoinEvent.EventID(), alice.UserID, rejoinEvent.PrevStateEventIDs(), rejoinEvent.PrevEventIDs()) + + state := currentRoomState(t, alice, room.RoomID) + mustHaveStateEventContent( + t, state, spec.MRoomName, "", "name", "after the leave", + "current state at rejoin has invalid room name event", + ) + mustHaveStateEventContent( + t, state, spec.MRoomMember, bob, "displayname", "bob after the leave 2", + "current state at rejoin has a stale membership event for the remote user", + ) + mustHaveStateEventContent( + t, state, spec.MRoomMember, alice.UserID, "membership", "join", + "the rejoining user is not joined", + ) + + // The room works after the rejoin: an event sent by the remote server arrives. + msg := srv.MustCreateEvent(t, room, federation.Event{ + Type: "m.room.message", + Sender: bob, + Content: map[string]interface{}{ + "msgtype": "m.text", + "body": "I am sent after the rejoin", + }, + }) + room.AddEvent(msg) + srv.MustSendTransaction(t, deployment, "hs1", []json.RawMessage{msg.JSON()}, nil) + alice.MustSyncUntil(t, client.SyncReq{}, client.SyncTimelineHasEventID(room.RoomID, msg.EventID())) +} + +// Test that a rejected event in the /send_join response, and the otherwise valid events which +// reference it via prev_state_events, are all rejected and so aren't part of the room's current state. +// +// MSC4242 specifies cascading rejection: "if A is rejected and B references A, then B is rejected +// and so on". A server which instead treats the state_dag as a flat set of state events (e.g. +// applying them in depth order) will pick up the rejected events and fail this test. +// +// The state DAG we build is: +// +// ALICE_LEAVE +// / \ +// TOPIC NAME <- both valid, both sent by bob +// | \ +// DORIS_NAME (rejected: doris is not in the room) +// | \ +// CHARLIE_JOIN (valid on its own, rejected for referencing DORIS_NAME) +// | \ +// BOB_NAME (valid on its own, rejected two hops from DORIS_NAME) +// \ +// ALICE_REJOIN, prev_state_events = [NAME] +// +// The rejected events are on a branch alongside the join: an event +// referencing a rejected event is itself rejected, so if the join could reach them the join would +// be rejected too. +func TestMSC4242JoinPublicRoomWithRejectedStateDAGEvents(t *testing.T) { + deployment := complement.Deploy(t, 1) + defer deployment.Destroy(t) + alice := deployment.Register(t, "hs1", helpers.RegistrationOpts{}) + + srv := federation.NewServer(t, deployment, + federation.HandleKeyRequests(), + federation.HandleTransactionRequests(nil, nil), + federation.HandleEventRequests(), + federation.HandleMakeSendJoinRequests(), + ) + srv.UnexpectedRequestsAreErrors = false + cancel := srv.Listen() + defer cancel() + + bob := srv.UserID("bob") + charlie := srv.UserID("charlie") + doris := srv.UserID("doris") + room := srv.MustMakeRoom(t, roomVersion, + federation.InitialRoomEvents(roomVersion, bob), + federation.WithImpl(ServerRoomImplStateDAG(t)), + ) + + // Join then leave, so that the fork point is a state event the homeserver already knows about. + alice.MustJoinRoom(t, room.RoomID, []spec.ServerName{srv.ServerName()}) + sinceJoined := alice.MustSyncUntil(t, client.SyncReq{}, client.SyncJoinedTo(alice.UserID, room.RoomID)) + alice.MustLeaveRoom(t, room.RoomID) + alice.MustSyncUntil(t, client.SyncReq{Since: sinceJoined}, client.SyncLeftFrom(alice.UserID, room.RoomID)) + leaveEvent := awaitMembership(t, room, alice.UserID, "leave") + + // Fork the state DAG at the leave event. + topic := mustCreateEvent(t, srv, room, MSC4242Event{ + Event: federation.Event{ + Type: spec.MRoomTopic, + Sender: bob, + StateKey: &empty, + Content: map[string]interface{}{"topic": "fork containing rejected events"}, + PrevEvents: []string{leaveEvent.EventID()}, + }, + PrevStateEvents: []string{leaveEvent.EventID()}, + }) + room.AddEvent(topic) + name := mustCreateEvent(t, srv, room, MSC4242Event{ + Event: federation.Event{ + Type: spec.MRoomName, + Sender: bob, + StateKey: &empty, + Content: map[string]interface{}{"name": "fork containing rejoin"}, + PrevEvents: []string{leaveEvent.EventID()}, + }, + PrevStateEvents: []string{leaveEvent.EventID()}, + }) + room.AddEvent(name) + + // Hang the rejected events off the topic event. Doris is not in the room, so her event fails + // auth and is rejected. + dorisName := mustCreateEvent(t, srv, room, MSC4242Event{ + Event: federation.Event{ + Type: spec.MRoomName, + Sender: doris, + StateKey: &empty, + Content: map[string]interface{}{"name": "doris is not in the room so this is rejected"}, + PrevEvents: []string{topic.EventID()}, + }, + PrevStateEvents: []string{topic.EventID()}, + }) + room.AddEvent(dorisName) + // Charlie's join would be allowed on its own as the room is public. + // It is rejected because it references a rejected event in prev_state_events. + charlieJoin := mustCreateEvent(t, srv, room, MSC4242Event{ + Event: federation.Event{ + Type: spec.MRoomMember, + Sender: charlie, + StateKey: &charlie, + Content: map[string]interface{}{"membership": spec.Join}, + PrevEvents: []string{dorisName.EventID()}, + }, + PrevStateEvents: []string{dorisName.EventID()}, + }) + room.AddEvent(charlieJoin) + // Bob may set the room name, and this event references a valid event, but it is rejected because + // that event is itself rejected two hops back. + bobName := mustCreateEvent(t, srv, room, MSC4242Event{ + Event: federation.Event{ + Type: spec.MRoomName, + Sender: bob, + StateKey: &empty, + Content: map[string]interface{}{"name": "rejected: two hops from a rejected event"}, + PrevEvents: []string{charlieJoin.EventID()}, + }, + PrevStateEvents: []string{charlieJoin.EventID()}, + }) + room.AddEvent(bobName) + + t.Logf( + "leave=%s topic=%s name=%s dorisName=%s charlieJoin=%s bobName=%s", + leaveEvent.EventID(), topic.EventID(), name.EventID(), + dorisName.EventID(), charlieJoin.EventID(), bobName.EventID(), + ) + + // Point the rejoin at the name event. + // ProtoEventCreatorFn will read this when it services the /make_join request and set this as prev_state_events + // but we return ALL state events in the room.Timeline when returning the state DAG in the /send_join response, + // meaning we will return the rejected fork. + room.ForwardExtremities = []string{name.EventID()} + + alice.MustJoinRoom(t, room.RoomID, []spec.ServerName{srv.ServerName()}) + alice.MustSyncUntil(t, client.SyncReq{}, client.SyncJoinedTo(alice.UserID, room.RoomID)) + + state := currentRoomState(t, alice, room.RoomID) + // Both forks should be in the current state + mustHaveStateEventContent( + t, state, "m.room.topic", "", "topic", "fork containing rejected events", + "the rejected fork was not merged into the current state", + ) + mustHaveStateEventContent( + t, state, spec.MRoomName, "", "name", "fork containing rejoin", + "current state not calculated correctly", + ) + mustNotHaveStateEvent( + t, state, spec.MRoomMember, charlie, + "charlie's join references a rejected event so must itself be rejected", + ) +} + +// setDisplayName sends numTimes membership events for userID which each change the display name, +// lengthening the state DAG. The events are added to the room but not sent anywhere: they are +// picked up by servers when they next join. +func setDisplayName(t *testing.T, srv *federation.Server, room *federation.ServerRoom, userID, prefix string, numTimes int) { + t.Helper() + for i := 0; i < numTimes; i++ { + time.Sleep(time.Millisecond) // ensure origin_server_ts changes + displayName := fmt.Sprintf("%s %d", prefix, i) + ev := mustAddStateEvent(t, srv, room, federation.Event{ + Type: spec.MRoomMember, + Sender: userID, + StateKey: &userID, + Content: map[string]interface{}{ + "membership": spec.Join, + "displayname": displayName, + }, + }) + t.Logf("state DAG: %s = (m.room.member, %s) displayname=%q prev_state_events=%v", + ev.EventID(), userID, displayName, ev.PrevStateEventIDs()) + } +} + +// mustSetRoomName adds an m.room.name to the Complement room and logs where it sits in the DAG. +func mustSetRoomName(t *testing.T, srv *federation.Server, room *federation.ServerRoom, sender, name string) gomatrixserverlib.PDU { + t.Helper() + ev := mustAddStateEvent(t, srv, room, federation.Event{ + Type: spec.MRoomName, + Sender: sender, + StateKey: &empty, + Content: map[string]interface{}{"name": name}, + }) + t.Logf("state DAG: %s = (m.room.name, \"\") name=%q prev_state_events=%v", + ev.EventID(), name, ev.PrevStateEventIDs()) + return ev +} diff --git a/tests/msc4242/state_dag_test.go b/tests/msc4242/state_dag_test.go new file mode 100644 index 000000000..da925b24f --- /dev/null +++ b/tests/msc4242/state_dag_test.go @@ -0,0 +1,254 @@ +package tests + +// Test harness for MSC4242 (State DAGs) + +import ( + "encoding/json" + "fmt" + "sort" + "strings" + "testing" + "time" + + "github.com/matrix-org/complement/client" + "github.com/matrix-org/complement/ct" + "github.com/matrix-org/complement/federation" + "github.com/matrix-org/complement/must" + "github.com/matrix-org/gomatrixserverlib" + "github.com/matrix-org/gomatrixserverlib/fclient" + "github.com/matrix-org/gomatrixserverlib/spec" + "github.com/tidwall/gjson" +) + +var ( + roomVersion gomatrixserverlib.RoomVersion = "org.matrix.msc4242.12" + empty = "" +) + +// ServerRoomImplStateDAG makes the room use state DAGs i.e set prev_state_events and generate valid +// /send_join responses. +func ServerRoomImplStateDAG(t ct.TestLike) federation.ServerRoomImpl { + findLastStateEventID := func(room *federation.ServerRoom) string { + room.TimelineMutex.RLock() + defer room.TimelineMutex.RUnlock() + for i := len(room.Timeline) - 1; i >= 0; i-- { + if room.Timeline[i].StateKey() != nil { + return room.Timeline[i].EventID() + } + } + t.Logf("%s: failed to find any state event in %d timeline events, no prev_state_events will be set!", room.RoomID, len(room.Timeline)) + return "" + } + return &federation.ServerRoomImplCustom{ + ServerRoomImplDefault: federation.ServerRoomImplDefault{}, + ProtoEventCreatorFn: func(def federation.ServerRoomImpl, room *federation.ServerRoom, ev federation.Event) (*gomatrixserverlib.ProtoEvent, error) { + proto, err := def.ProtoEventCreator(room, ev) + if err != nil { + return nil, err + } + proto.AuthEvents = nil + + if ev.Type == spec.MRoomCreate && ev.StateKey != nil && *ev.StateKey == "" { + proto.PrevStateEvents = &[]string{} + } else { + // if the fwd extrems are state, use that. + var fwdExtrems []gomatrixserverlib.PDU + for _, id := range room.ForwardExtremities { + pdu, ok := room.GetEventInTimeline(id) + if ok && pdu.StateKey() != nil { + fwdExtrems = append(fwdExtrems, pdu) + } + } + if len(fwdExtrems) == 0 { + proto.PrevStateEvents = &[]string{ + findLastStateEventID(room), + } + } else { + ids := make([]string, len(fwdExtrems)) + for i := range ids { + ids[i] = fwdExtrems[i].EventID() + } + proto.PrevStateEvents = &ids + } + } + return proto, nil + }, + PopulateFromSendJoinResponseFn: func(def federation.ServerRoomImpl, room *federation.ServerRoom, joinEvent gomatrixserverlib.PDU, resp fclient.RespSendJoin) { + stateDAGEvents := resp.StateDAG.UntrustedEvents(roomVersion) + sort.Slice(stateDAGEvents, func(i, j int) bool { + return stateDAGEvents[i].Depth() < stateDAGEvents[j].Depth() + }) + // we assume no forks and no rejected events, so we can just bluntly replace events in + // depth order to work out the current state + for _, state := range stateDAGEvents { + room.ReplaceCurrentState(state) + } + room.AddEvent(joinEvent) + }, + GenerateSendJoinResponseFn: func(def federation.ServerRoomImpl, room *federation.ServerRoom, s *federation.Server, joinEvent gomatrixserverlib.PDU, expectPartialState, omitServersInRoom bool) fclient.RespSendJoin { + res := fclient.RespSendJoin{ + ServersInRoom: []string{}, + } + res.Event = joinEvent.JSON() + res.MembersOmitted = omitServersInRoom + for _, ev := range room.Timeline { + if ev.StateKey() != nil { + res.StateDAG = append(res.StateDAG, ev.JSON()) + } + } + for _, srvName := range room.ServersInRoom() { + res.ServersInRoom = append(res.ServersInRoom, string(srvName)) + } + room.AddEvent(joinEvent) + return res + }, + } +} + +// MSC4242Event is a federation.Event with explicitly set prev_state_events. Use this when you need +// to control the shape of the state DAG, e.g to create forks. Events made with +// Server.MustCreateEvent instead reference the room's current state DAG extremities. +type MSC4242Event struct { + federation.Event + PrevStateEvents []string +} + +// mustCreateEvent creates and signs an event with explicitly set prev_state_events. It does not add +// the event to the room: see ServerRoom.AddEvent for that. +func mustCreateEvent(t ct.TestLike, s *federation.Server, room *federation.ServerRoom, ev MSC4242Event) gomatrixserverlib.PDU { + t.Helper() + content, err := json.Marshal(ev.Content) + if err != nil { + ct.Fatalf(t, "mustCreateEvent: failed to marshal event content %s - %+v", err, ev.Content) + } + var unsigned []byte + if ev.Unsigned != nil { + unsigned, err = json.Marshal(ev.Unsigned) + if err != nil { + ct.Fatalf(t, "mustCreateEvent: failed to marshal event unsigned: %s - %+v", err, ev.Unsigned) + } + } + + var prevEvents interface{} + if ev.PrevEvents != nil { + // We deliberately want to set the prev events. + prevEvents = ev.PrevEvents + } else { + // No other prev events were supplied so we'll just + // use the forward extremities of the room, which is + // the usual behaviour. + prevEvents = room.ForwardExtremities + } + proto := gomatrixserverlib.ProtoEvent{ + SenderID: ev.Sender, + Depth: int64(room.Depth + 1), // depth starts at 1 + Type: ev.Type, + StateKey: ev.StateKey, + Content: content, + RoomID: room.RoomID, + PrevEvents: prevEvents, + Unsigned: unsigned, + Redacts: ev.Redacts, + PrevStateEvents: &ev.PrevStateEvents, + } + // auth_events do not exist on state DAG events: they are calculated from prev_state_events by + // the receiving server, and the event builder strips the field. + signedEvent, err := room.EventCreator(room, s, &proto) + if err != nil { + ct.Fatalf(t, "mustCreateEvent: failed to create event: %s", err) + } + return signedEvent +} + +// mustAddStateEvent creates a state event whose prev_state_events are the room's current state DAG +// extremities, then adds it to the room, making it the sole extremity. +func mustAddStateEvent(t *testing.T, srv *federation.Server, room *federation.ServerRoom, ev federation.Event) gomatrixserverlib.PDU { + t.Helper() + pdu := srv.MustCreateEvent(t, room, ev) + room.AddEvent(pdu) + return pdu +} + +// awaitMembership blocks until the Complement server has been told about a membership change for +// userID via a /send transaction, returning the membership event. +// +// Homeservers create leave events for their own joined users locally and federate them in a +// transaction, rather than calling /make_leave, so we have to wait for the event to arrive before +// building on top of it. +func awaitMembership(t *testing.T, room *federation.ServerRoom, userID, membership string) gomatrixserverlib.PDU { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for { + ev := room.CurrentState(spec.MRoomMember, userID) + if ev != nil && gjson.GetBytes(ev.Content(), "membership").Str == membership { + return ev + } + if time.Now().After(deadline) { + got := "" + if ev != nil { + got = gjson.GetBytes(ev.Content(), "membership").Str + } + ct.Fatalf(t, "awaitMembership: timed out waiting for %s to be '%s' in %s, got '%s'", + userID, membership, room.RoomID, got) + } + time.Sleep(50 * time.Millisecond) + } +} + +// currentRoomState returns the current room state as seen by the client, keyed off (type, state_key). +func currentRoomState(t *testing.T, cli *client.CSAPI, roomID string) map[[2]string]gjson.Result { + t.Helper() + res := cli.MustDo(t, "GET", []string{"_matrix", "client", "v3", "rooms", roomID, "state"}) + state := make(map[[2]string]gjson.Result) + for _, ev := range must.ParseJSON(t, res.Body).Array() { + state[[2]string{ev.Get("type").Str, ev.Get("state_key").Str}] = ev + } + return state +} + +// mustNotHaveStateEvent fails the test if the given state tuple is present in the room state. +func mustNotHaveStateEvent(t *testing.T, state map[[2]string]gjson.Result, evType, stateKey, reason string) { + t.Helper() + if ev, ok := state[[2]string{evType, stateKey}]; ok { + ct.Fatalf(t, "room state unexpectedly contains (%s, %s): %s\n%s\n%s", + evType, stateKey, reason, ev.Raw, formatState(state)) + } +} + +// mustHaveStateEventContent fails the test unless the given state tuple is present and the given +// content field matches. +func mustHaveStateEventContent(t *testing.T, state map[[2]string]gjson.Result, evType, stateKey, field, want, reason string) { + t.Helper() + ev, ok := state[[2]string{evType, stateKey}] + if !ok { + ct.Fatalf(t, "room state is missing (%s, %s): %s\n%s", evType, stateKey, reason, formatState(state)) + } + got := ev.Get("content." + field).Str + if got != want { + ct.Fatalf(t, "room state (%s, %s) content.%s: got '%s' want '%s': %s\n%s", + evType, stateKey, field, got, want, reason, formatState(state)) + } +} + +// formatState renders the whole room state so a failed assertion shows what the server actually +// resolved, not just the tuple which was checked. +func formatState(state map[[2]string]gjson.Result) string { + keys := make([][2]string, 0, len(state)) + for k := range state { + keys = append(keys, k) + } + sort.Slice(keys, func(i, j int) bool { + if keys[i][0] != keys[j][0] { + return keys[i][0] < keys[j][0] + } + return keys[i][1] < keys[j][1] + }) + var sb strings.Builder + sb.WriteString("full room state as the server resolved it:\n") + for _, k := range keys { + ev := state[k] + fmt.Fprintf(&sb, " (%s, %s) %s content=%s\n", + k[0], k[1], ev.Get("event_id").Str, ev.Get("content").Raw) + } + return sb.String() +}