From 0f3c1860fa0b7deb42ae28660dfea2522a2f74b0 Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Mon, 22 Jun 2026 22:57:49 +0100 Subject: [PATCH] Fix racy internal test One of the internal tests relies on receiving all the events from sliding sync. However, this doens't actually happen unless you set up a subscription to the room (otherwise, a timeline limit of 1 is applied, which means that if you don't sync fast enough you can miss an event). So, add the infrastructure for setting up a subscription and use it. Fixes: https://github.com/matrix-org/complement-crypto/issues/246 --- internal/api/client.go | 5 +++++ internal/api/js/js.go | 5 +++++ internal/api/rust/rust.go | 15 +++++++++++++++ internal/deploy/rpc/client.go | 9 +++++++++ internal/deploy/rpc/server.go | 5 +++++ internal/tests/client_test.go | 15 +++++++++++---- 6 files changed, 50 insertions(+), 4 deletions(-) diff --git a/internal/api/client.go b/internal/api/client.go index cbeb8fd..1433733 100644 --- a/internal/api/client.go +++ b/internal/api/client.go @@ -39,6 +39,11 @@ type Client interface { // MUST BLOCK until the initial sync is complete. // Returns an error if there was a problem syncing. StartSyncing(t ct.TestLike) (stopSyncing func(), err error) + // SubscribeToRoom sets up a subscription to the given room, on SDKs that use sliding sync. (On SDKs that + // use regular sync, this is a no-op.) + // + // StartSyncing must have been called before this function. + SubscribeToRoom(t ct.TestLike, roomId string) error // IsRoomEncrypted returns true if the room is encrypted. May return an error e.g if you // provide a bogus room ID. IsRoomEncrypted(t ct.TestLike, roomID string) (bool, error) diff --git a/internal/api/js/js.go b/internal/api/js/js.go index f5c342b..6a25f8f 100644 --- a/internal/api/js/js.go +++ b/internal/api/js/js.go @@ -670,6 +670,11 @@ func (c *JSClient) StartSyncing(t ct.TestLike) (stopSyncing func(), err error) { }, nil } +// SubscribeToRoom is a no-op on the js sdk, because it uses regular sync. +func (c *JSClient) SubscribeToRoom(t ct.TestLike, roomID string) error { + return nil +} + // IsRoomEncrypted returns true if the room is encrypted. May return an error e.g if you // provide a bogus room ID. func (c *JSClient) IsRoomEncrypted(t ct.TestLike, roomID string) (bool, error) { diff --git a/internal/api/rust/rust.go b/internal/api/rust/rust.go index 2247ba4..08ba3f4 100644 --- a/internal/api/rust/rust.go +++ b/internal/api/rust/rust.go @@ -531,6 +531,21 @@ func (c *RustClient) StartSyncing(t ct.TestLike) (stopSyncing func(), err error) }, nil } +// SubscribeToRoom sets up a subscription to the given room, on SDKs that use sliding sync. (On SDKs that +// use regular sync, this is a no-op.) +// +// StartSyncing must have been called before this function. +func (c *RustClient) SubscribeToRoom(t ct.TestLike, roomID string) error { + t.Helper() + if c.syncService == nil { + return fmt.Errorf("cannot subscribe to room %s: StartSyncing not yet called", roomID) + } + if err := c.syncService.RoomListService().SubscribeToRooms([]string{roomID}); err != nil { + return fmt.Errorf("cannot subscribe to room %s: %s", roomID, err) + } + return nil +} + // IsRoomEncrypted returns true if the room is encrypted. May return an error e.g if you // provide a bogus room ID. func (c *RustClient) IsRoomEncrypted(t ct.TestLike, roomID string) (bool, error) { diff --git a/internal/deploy/rpc/client.go b/internal/deploy/rpc/client.go index e2ed338..320dedd 100644 --- a/internal/deploy/rpc/client.go +++ b/internal/deploy/rpc/client.go @@ -236,6 +236,15 @@ func (c *RPCClient) StartSyncing(t ct.TestLike) (stopSyncing func(), err error) }, nil } +// SubscribeToRoom sets up a subscription to the given room, on SDKs that use sliding sync. (On SDKs that +// use regular sync, this is a no-op.) +// +// StartSyncing must have been called before this function. +func (c *RPCClient) SubscribeToRoom(t ct.TestLike, roomID string) error { + var void int + return c.client.Call("Server.SubscribeToRoom", roomID, &void) +} + // IsRoomEncrypted returns true if the room is encrypted. May return an error e.g if you // provide a bogus room ID. func (c *RPCClient) IsRoomEncrypted(t ct.TestLike, roomID string) (bool, error) { diff --git a/internal/deploy/rpc/server.go b/internal/deploy/rpc/server.go index afe6bc3..b0ade06 100644 --- a/internal/deploy/rpc/server.go +++ b/internal/deploy/rpc/server.go @@ -131,6 +131,11 @@ func (s *Server) StopSyncing(testName string, void *int) error { return nil } +func (s *Server) SubscribeToRoom(roomID string, void *int) error { + defer s.keepAlive() + return s.activeClient.SubscribeToRoom(&api.MockT{}, roomID) +} + func (s *Server) IsRoomEncrypted(roomID string, isEncrypted *bool) error { defer s.keepAlive() var err error diff --git a/internal/tests/client_test.go b/internal/tests/client_test.go index bf4ed13..37fc896 100644 --- a/internal/tests/client_test.go +++ b/internal/tests/client_test.go @@ -90,8 +90,7 @@ func TestMain(m *testing.M) { func TestReceiveTimeline(t *testing.T) { deployment := Deploy(t) - createAndSendEvents := func(t *testing.T, csapi *client.CSAPI) (roomID string, eventIDs []string) { - roomID = csapi.MustCreateRoom(t, map[string]interface{}{}) + createAndSendEvents := func(t *testing.T, csapi *client.CSAPI, roomID string) (eventIDs []string) { for i := 0; i < 10; i++ { eventIDs = append(eventIDs, csapi.SendEventSynced(t, roomID, b.Event{ Type: "m.room.message", @@ -107,7 +106,8 @@ func TestReceiveTimeline(t *testing.T) { // test that if we start syncing with a room full of events, we see those events. ForEachClient(t, "existing_events", deployment, func(t *testing.T, client api.TestClient, csapi *client.CSAPI) { must.NotError(t, "Failed to login", client.Login(t, client.Opts())) - roomID, eventIDs := createAndSendEvents(t, csapi) + roomID := csapi.MustCreateRoom(t, map[string]interface{}{}) + eventIDs := createAndSendEvents(t, csapi, roomID) time.Sleep(time.Second) // give time for everything to settle server-side e.g sliding sync proxy stopSyncing := client.MustStartSyncing(t) defer stopSyncing() @@ -134,17 +134,24 @@ func TestReceiveTimeline(t *testing.T) { // test that if we are already syncing and then see a room live stream full of events, we see those events. ForEachClient(t, "live_events", deployment, func(t *testing.T, client api.TestClient, csapi *client.CSAPI) { + roomID := csapi.MustCreateRoom(t, map[string]interface{}{}) + must.NotError(t, "Failed to login", client.Login(t, client.Opts())) stopSyncing := client.MustStartSyncing(t) defer stopSyncing() + // Subscribe to the room, so that sliding sync returns all events. + must.NotError(t,"could not subscribe to room", client.SubscribeToRoom(t, roomID)) + time.Sleep(time.Second) // give time for syncing to be well established. // send the messages whilst syncing. - roomID, eventIDs := createAndSendEvents(t, csapi) + eventIDs := createAndSendEvents(t, csapi, roomID) + // ensure we see all the events for i, eventID := range eventIDs { t.Logf("waiting for event %d : %s", i, eventID) client.WaitUntilEventInRoom(t, roomID, api.CheckEventHasEventID(eventID)).Waitf(t, 5*time.Second, "client did not see event %s", eventID) } + // now send another live event and ensure we see it. This ensure we can still wait for events after having // previously waited for events. waiter := client.WaitUntilEventInRoom(t, roomID, api.CheckEventHasBody("Final"))