Fix racy internal test

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)

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) {

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) {

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) {

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

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"))