Draft: Test that /messages works on remote homeserver and can be backfilled properly after many batches (MSC2716)

internal/client/client.go

@@ -310,7 +310,7 @@ func (c *CSAPI) DoFunc(t *testing.T, method string, paths []string, opts ...Requ
 	}
 	// debug log the request
 	if c.Debug {
-		t.Logf("Making %s request to %s", method, reqURL)
+		t.Logf("Making %s request to %s", method, req.URL)
 		contentType := req.Header.Get("Content-Type")
 		if contentType == "application/json" || strings.HasPrefix(contentType, "text/") {
 			if req.Body != nil {

tests/msc2716_test.go

@@ -49,18 +49,19 @@ var (
 var createPublicRoomOpts = map[string]interface{}{
 	"preset":       "public_chat",
 	"name":         "the hangout spot",
-	"room_version": "org.matrix.msc2716v3",
+	"room_version": "org.matrix.msc2716v4",
 }
 
 var createPrivateRoomOpts = map[string]interface{}{
 	"preset":       "private_chat",
 	"name":         "the hangout spot",
-	"room_version": "org.matrix.msc2716v3",
+	"room_version": "org.matrix.msc2716v4",
 }
 
 func TestImportHistoricalMessages(t *testing.T) {
 	deployment := Deploy(t, b.BlueprintHSWithApplicationService)
 	defer deployment.Destroy(t)
+	//defer time.Sleep(2 * time.Hour)
 
 	// Create the application service bridge user that is able to import historical messages
 	asUserID := "@the-bridge-user:hs1"
@@ -98,7 +99,7 @@ func TestImportHistoricalMessages(t *testing.T) {
 			//
 			// Create the first batch including the "live" event we are going to
 			// import our historical events next to.
-			eventIDsBefore := createMessagesInRoom(t, alice, roomID, 2)
+			eventIDsBefore := createMessagesInRoom(t, alice, roomID, 2, "eventIDsBefore")
 			eventIdBefore := eventIDsBefore[len(eventIDsBefore)-1]
 			timeAfterEventBefore := time.Now()
 
@@ -110,7 +111,7 @@ func TestImportHistoricalMessages(t *testing.T) {
 			// Create the second batch of events.
 			// This will also fill up the buffer so we have to scrollback to the
 			// inserted history later.
-			eventIDsAfter := createMessagesInRoom(t, alice, roomID, 2)
+			eventIDsAfter := createMessagesInRoom(t, alice, roomID, 2, "eventIDsAfter")
 
 			// Insert the most recent batch of historical messages
 			insertTime0 := timeAfterEventBefore.Add(timeBetweenMessages * 3)
@@ -214,7 +215,7 @@ func TestImportHistoricalMessages(t *testing.T) {
 			alice.JoinRoom(t, roomID, nil)
 
 			// Create the "live" event we are going to insert our historical events next to
-			eventIDsBefore := createMessagesInRoom(t, alice, roomID, 1)
+			eventIDsBefore := createMessagesInRoom(t, alice, roomID, 1, "eventIDsBefore")
 			eventIdBefore := eventIDsBefore[0]
 			timeAfterEventBefore := time.Now()
 
@@ -255,19 +256,74 @@ func TestImportHistoricalMessages(t *testing.T) {
 			})
 		})
 
+		t.Run("Backfill still works after many batches are imported", func(t *testing.T) {
+			t.Parallel()
+
+			roomID := as.CreateRoom(t, createPublicRoomOpts)
+			alice.JoinRoom(t, roomID, nil)
+
+			// Create some normal messages in the timeline
+			eventIDsBefore := createMessagesInRoom(t, alice, roomID, 2, "eventIDsBefore")
+			eventIdBefore := eventIDsBefore[len(eventIDsBefore)-1]
+			timeAfterEventBefore := time.Now()
+
+			// wait X number of ms to ensure that the timestamp changes enough for
+			// each of the historical messages we try to import later
+			//numBatches := 11
+			numBatches := 2
+			numHistoricalMessagesPerBatch := 100
+			time.Sleep(time.Duration(numBatches*numHistoricalMessagesPerBatch) * timeBetweenMessages)
+
+			// eventIDsAfter
+			createMessagesInRoom(t, alice, roomID, 2, "eventIDsAfter")
+
+			// Import a long chain of batches connected to each other.
+			// We want to make sure Synapse doesn't blow up after we import
+			// many messages.
+			var expectedEventIDs []string
+			nextBatchID := ""
+			for i := 0; i < numBatches; i++ {
+				insertTime := timeAfterEventBefore.Add(timeBetweenMessages * time.Duration(numBatches-numHistoricalMessagesPerBatch*i))
+				batchSendRes := batchSendHistoricalMessages(
+					t,
+					as,
+					roomID,
+					eventIdBefore,
+					nextBatchID,
+					createJoinStateEventsForBatchSendRequest([]string{virtualUserID}, insertTime),
+					createMessageEventsForBatchSendRequest([]string{virtualUserID}, insertTime, numHistoricalMessagesPerBatch),
+					// Status
+					200,
+				)
+				batchSendResBody := client.ParseJSON(t, batchSendRes)
+				// Make sure we see all of the historical messages
+				expectedEventIDs = append(expectedEventIDs, client.GetJSONFieldStringArray(t, batchSendResBody, "event_ids")...)
+				nextBatchID = client.GetJSONFieldStr(t, batchSendResBody, "next_batch_id")
+			}
+
+			// Make sure we see the events at the very start of the message history
+			expectedEventIDs = append(expectedEventIDs, eventIDsBefore...)
+
+			// Join the room from a remote homeserver after the historical messages were sent
+			remoteCharlie.JoinRoom(t, roomID, []string{"hs1"})
+
+			// Make sure events can be backfilled from the remote homeserver
+			paginateUntilMessageCheckOff(t, remoteCharlie, roomID, expectedEventIDs)
+		})
+
 		t.Run("Historical events from /batch_send do not come down in an incremental sync", func(t *testing.T) {
 			t.Parallel()
 
 			roomID := as.CreateRoom(t, createPublicRoomOpts)
 			alice.JoinRoom(t, roomID, nil)
 
 			// Create the "live" event we are going to insert our historical events next to
-			eventIDsBefore := createMessagesInRoom(t, alice, roomID, 1)
+			eventIDsBefore := createMessagesInRoom(t, alice, roomID, 1, "eventIDsBefore")
 			eventIdBefore := eventIDsBefore[0]
 			timeAfterEventBefore := time.Now()
 
 			// Create some "live" events to saturate and fill up the /sync response
-			createMessagesInRoom(t, alice, roomID, 5)
+			createMessagesInRoom(t, alice, roomID, 5, "live")
 
 			// Import a historical event
 			batchSendRes := batchSendHistoricalMessages(
@@ -286,7 +342,7 @@ func TestImportHistoricalMessages(t *testing.T) {
 			historicalEventId := historicalEventIDs[0]
 
 			// This is just a dummy event we search for after the historicalEventId
-			eventIDsAfterHistoricalImport := createMessagesInRoom(t, alice, roomID, 1)
+			eventIDsAfterHistoricalImport := createMessagesInRoom(t, alice, roomID, 1, "eventIDsAfterHistoricalImport")
 			eventIDAfterHistoricalImport := eventIDsAfterHistoricalImport[0]
 
 			// Sync until we find the eventIDAfterHistoricalImport.
@@ -309,7 +365,7 @@ func TestImportHistoricalMessages(t *testing.T) {
 			alice.JoinRoom(t, roomID, nil)
 
 			// Create the "live" event we are going to import our historical events next to
-			eventIDsBefore := createMessagesInRoom(t, alice, roomID, 1)
+			eventIDsBefore := createMessagesInRoom(t, alice, roomID, 1, "eventIDsBefore")
 			eventIdBefore := eventIDsBefore[0]
 			timeAfterEventBefore := time.Now()
 
@@ -363,7 +419,7 @@ func TestImportHistoricalMessages(t *testing.T) {
 			roomID := as.CreateRoom(t, createPublicRoomOpts)
 			alice.JoinRoom(t, roomID, nil)
 
-			eventIDsBefore := createMessagesInRoom(t, alice, roomID, 1)
+			eventIDsBefore := createMessagesInRoom(t, alice, roomID, 1, "eventIDsBefore")
 			eventIdBefore := eventIDsBefore[0]
 			timeAfterEventBefore := time.Now()
 
@@ -386,7 +442,7 @@ func TestImportHistoricalMessages(t *testing.T) {
 			roomID := as.CreateRoom(t, createPublicRoomOpts)
 			alice.JoinRoom(t, roomID, nil)
 
-			eventIDsBefore := createMessagesInRoom(t, alice, roomID, 1)
+			eventIDsBefore := createMessagesInRoom(t, alice, roomID, 1, "eventIDsBefore")
 			eventIdBefore := eventIDsBefore[0]
 			timeAfterEventBefore := time.Now()
 
@@ -417,7 +473,7 @@ func TestImportHistoricalMessages(t *testing.T) {
 			alice.JoinRoom(t, roomID, nil)
 
 			// Create the "live" event we are going to import our historical events next to
-			eventIDsBefore := createMessagesInRoom(t, alice, roomID, 1)
+			eventIDsBefore := createMessagesInRoom(t, alice, roomID, 1, "eventIDsBefore")
 			eventIdBefore := eventIDsBefore[0]
 			timeAfterEventBefore := time.Now()
 
@@ -469,12 +525,12 @@ func TestImportHistoricalMessages(t *testing.T) {
 			roomID := as.CreateRoom(t, createPublicRoomOpts)
 			alice.JoinRoom(t, roomID, nil)
 
-			eventIDsBefore := createMessagesInRoom(t, alice, roomID, 1)
+			eventIDsBefore := createMessagesInRoom(t, alice, roomID, 1, "eventIDsBefore")
 			eventIdBefore := eventIDsBefore[0]
 			timeAfterEventBefore := time.Now()
 
 			// eventIDsAfter
-			createMessagesInRoom(t, alice, roomID, 3)
+			createMessagesInRoom(t, alice, roomID, 3, "eventIDsAfter")
 
 			batchSendRes := batchSendHistoricalMessages(
 				t,
@@ -522,7 +578,7 @@ func TestImportHistoricalMessages(t *testing.T) {
 			roomID := as.CreateRoom(t, createPublicRoomOpts)
 			alice.JoinRoom(t, roomID, nil)
 
-			eventIDsBefore := createMessagesInRoom(t, alice, roomID, 1)
+			eventIDsBefore := createMessagesInRoom(t, alice, roomID, 1, "eventIDsBefore")
 			eventIdBefore := eventIDsBefore[0]
 			timeAfterEventBefore := time.Now()
 
@@ -546,7 +602,7 @@ func TestImportHistoricalMessages(t *testing.T) {
 			})
 
 			// eventIDsAfter
-			createMessagesInRoom(t, alice, roomID, 3)
+			createMessagesInRoom(t, alice, roomID, 3, "eventIDsAfter")
 
 			batchSendRes := batchSendHistoricalMessages(
 				t,
@@ -597,12 +653,12 @@ func TestImportHistoricalMessages(t *testing.T) {
 			// Join the room from a remote homeserver before any historical messages are sent
 			remoteCharlie.JoinRoom(t, roomID, []string{"hs1"})
 
-			eventIDsBefore := createMessagesInRoom(t, alice, roomID, 1)
+			eventIDsBefore := createMessagesInRoom(t, alice, roomID, 1, "eventIDsBefore")
 			eventIdBefore := eventIDsBefore[0]
 			timeAfterEventBefore := time.Now()
 
 			// eventIDsAfter
-			createMessagesInRoom(t, alice, roomID, 10)
+			createMessagesInRoom(t, alice, roomID, 10, "eventIDsAfter")
 
 			// Mimic scrollback just through the latest messages
 			remoteCharlie.MustDoFunc(t, "GET", []string{"_matrix", "client", "r0", "rooms", roomID, "messages"}, client.WithContentType("application/json"), client.WithQueries(url.Values{
@@ -685,12 +741,12 @@ func TestImportHistoricalMessages(t *testing.T) {
 			// Join the room from a remote homeserver before any historical messages are sent
 			remoteCharlie.JoinRoom(t, roomID, []string{"hs1"})
 
-			eventIDsBefore := createMessagesInRoom(t, alice, roomID, 1)
+			eventIDsBefore := createMessagesInRoom(t, alice, roomID, 1, "eventIDsBefore")
 			eventIdBefore := eventIDsBefore[0]
 			timeAfterEventBefore := time.Now()
 
 			// eventIDsAfter
-			createMessagesInRoom(t, alice, roomID, 3)
+			createMessagesInRoom(t, alice, roomID, 3, "eventIDsAfter")
 
 			// Mimic scrollback to all of the messages
 			// scrollbackMessagesRes
@@ -778,7 +834,7 @@ func TestImportHistoricalMessages(t *testing.T) {
 				alice.JoinRoom(t, roomID, nil)
 
 				// Create the "live" event we are going to import our historical events next to
-				eventIDsBefore := createMessagesInRoom(t, alice, roomID, 1)
+				eventIDsBefore := createMessagesInRoom(t, alice, roomID, 1, "eventIDsBefore")
 				eventIdBefore := eventIDsBefore[0]
 				timeAfterEventBefore := time.Now()
 
@@ -818,7 +874,7 @@ func TestImportHistoricalMessages(t *testing.T) {
 				alice.JoinRoom(t, roomID, nil)
 
 				// Create the "live" event we are going to import our historical events next to
-				eventIDsBefore := createMessagesInRoom(t, alice, roomID, 1)
+				eventIDsBefore := createMessagesInRoom(t, alice, roomID, 1, "eventIDsBefore")
 				eventIdBefore := eventIDsBefore[0]
 				timeAfterEventBefore := time.Now()
 
@@ -913,6 +969,96 @@ func fetchUntilMessagesResponseHas(t *testing.T, c *client.CSAPI, roomID string,
 	}
 }
 
+func paginateUntilMessageCheckOff(t *testing.T, c *client.CSAPI, roomID string, expectedEventIDs []string) {
+	t.Helper()
+	start := time.Now()
+
+	workingExpectedEventIDMap := make(map[string]string)
+	for _, expectedEventID := range expectedEventIDs {
+		workingExpectedEventIDMap[expectedEventID] = expectedEventID
+	}
+
+	var actualEventIDList []string
+	checkCounter := 0
+	messageResEnd := ""
+	generateErrorMesssageInfo := func() string {
+		i := 0
+		leftoverEventIDs := make([]string, len(workingExpectedEventIDMap))
+		for eventID := range workingExpectedEventIDMap {
+			leftoverEventIDs[i] = eventID
+			i++
+		}
+
+		return fmt.Sprintf("Called /messages %d times but only found %d/%d expected messages. Leftover messages we expected (%d): %s. We saw %d events over all of the API calls: %s",
+			checkCounter,
+			len(expectedEventIDs)-len(leftoverEventIDs),
+			len(expectedEventIDs),
+			len(leftoverEventIDs),
+			leftoverEventIDs,
+			len(actualEventIDList),
+			actualEventIDList,
+		)
+	}
+
+	for {
+		if time.Since(start) > 200*c.SyncUntilTimeout {
+			t.Fatalf(
+				"paginateUntilMessageCheckOff timed out. %s",
+				generateErrorMesssageInfo(),
+			)
+		}
+
+		messagesRes := c.MustDoFunc(t, "GET", []string{"_matrix", "client", "r0", "rooms", roomID, "messages"}, client.WithContentType("application/json"), client.WithQueries(url.Values{
+			"dir": []string{"b"},
+			// TODO: Can we do it with 100?
+			"limit": []string{"100"},
+			"from":  []string{messageResEnd},
+		}))
+		messsageResBody := client.ParseJSON(t, messagesRes)
+
+		messageResEnd = client.GetJSONFieldStr(t, messsageResBody, "end")
+
+		wantKey := "chunk"
+		keyRes := gjson.GetBytes(messsageResBody, wantKey)
+		if !keyRes.Exists() {
+			t.Fatalf("missing key '%s'", wantKey)
+		}
+		if !keyRes.IsArray() {
+			t.Fatalf("key '%s' is not an array (was %s)", wantKey, keyRes.Type)
+		}
+
+		events := keyRes.Array()
+
+		if len(events) == 0 {
+			t.Fatalf(
+				"paginateUntilMessageCheckOff reached the end of the messages without finding all expected events. %s",
+				generateErrorMesssageInfo(),
+			)
+		}
+
+		// logrus.WithFields(logrus.Fields{
+		// 	"events":        events,
+		// 	"messageResEnd": messageResEnd,
+		// }).Error("paginateUntilMessageCheckOff chunk")
+		for _, ev := range events {
+			eventID := ev.Get("event_id").Str
+			actualEventIDList = append(actualEventIDList, eventID)
+
+			if _, keyExists := workingExpectedEventIDMap[eventID]; keyExists {
+				delete(workingExpectedEventIDMap, eventID)
+			}
+
+			if len(workingExpectedEventIDMap) == 0 {
+				return
+			}
+		}
+
+		checkCounter++
+		// Add a slight delay so we don't hammmer the messages endpoint
+		time.Sleep(500 * time.Millisecond)
+	}
+}
+
 func isRelevantEvent(r gjson.Result) bool {
 	return len(r.Get("content").Get("body").Str) > 0 ||
 		r.Get("type").Str == insertionEventType ||
@@ -1003,14 +1149,14 @@ func sendMarkerAndEnsureBackfilled(t *testing.T, as *client.CSAPI, c *client.CSA
 	return markerEventID
 }
 
-func createMessagesInRoom(t *testing.T, c *client.CSAPI, roomID string, count int) (eventIDs []string) {
+func createMessagesInRoom(t *testing.T, c *client.CSAPI, roomID string, count int, messageSuffix string) (eventIDs []string) {
 	eventIDs = make([]string, count)
 	for i := 0; i < len(eventIDs); i++ {
 		newEvent := b.Event{
 			Type: "m.room.message",
 			Content: map[string]interface{}{
 				"msgtype": "m.text",
-				"body":    fmt.Sprintf("Message %d", i),
+				"body":    fmt.Sprintf("Message %d (%s)", i, messageSuffix),
 			},
 		}
 		newEventId := c.SendEventSynced(t, roomID, newEvent)