Skip to content

Marker events as state - MSC2716 #371

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 9 commits into from
May 24, 2022
Merged

Marker events as state - MSC2716 #371

Changes from 5 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
056ebc0
Marker events as state
MadLittleMods May 7, 2022
17e265a
Barebones test to reproduce "Current state for room {room_id} is empt…
MadLittleMods May 12, 2022
1171ed9
Revert "Barebones test to reproduce "Current state for room {room_id}…
MadLittleMods May 12, 2022
b53ec21
Better comments
MadLittleMods May 12, 2022
3b3d383
Add test for multiple marker state
MadLittleMods May 12, 2022
77559bf
Marker events should have unique state_key so they all show up in the…
MadLittleMods May 19, 2022
3ac5a05
Test can be made generic between these two cases
MadLittleMods May 19, 2022
1ff578b
Make comment accurate
MadLittleMods May 19, 2022
73c78cf
Remove vestigial timeout from testing in another PR
MadLittleMods May 20, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
269 changes: 267 additions & 2 deletions tests/msc2716_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -918,6 +918,175 @@ func TestImportHistoricalMessages(t *testing.T) {
},
})
})

t.Run("Historical messages show up for remote federated homeserver even when the homeserver is missing the part of the timeline where the marker was sent and it paginates before it occured", func(t *testing.T) {
t.Parallel()

roomID := as.CreateRoom(t, createPublicRoomOpts)
alice.JoinRoom(t, roomID, nil)

eventIDsBefore := createMessagesInRoom(t, alice, roomID, 1, "eventIDsBefore")
eventIdBefore := eventIDsBefore[0]
timeAfterEventBefore := time.Now()

eventIDsAfter := createMessagesInRoom(t, alice, roomID, 3, "eventIDsAfter")
eventIDAfter := eventIDsAfter[0]

// Join the room from a remote homeserver before the historical messages were sent
remoteCharlie.JoinRoom(t, roomID, []string{"hs1"})

// Make sure all of the events have been backfilled for the remote user
// before we leave the room
fetchUntilMessagesResponseHas(t, remoteCharlie, roomID, func(ev gjson.Result) bool {
if ev.Get("event_id").Str == eventIdBefore {
return true
}

return false
})

// Leave before the historical messages are imported
remoteCharlie.LeaveRoom(t, roomID)

batchSendRes := batchSendHistoricalMessages(
t,
as,
roomID,
eventIdBefore,
"",
createJoinStateEventsForBatchSendRequest([]string{virtualUserID}, timeAfterEventBefore),
createMessageEventsForBatchSendRequest([]string{virtualUserID}, timeAfterEventBefore, 2),
// Status
200,
)
batchSendResBody := client.ParseJSON(t, batchSendRes)
historicalEventIDs := client.GetJSONFieldStringArray(t, batchSendResBody, "event_ids")
baseInsertionEventID := client.GetJSONFieldStr(t, batchSendResBody, "base_insertion_event_id")

// Send the marker event which lets remote homeservers know there are
// some historical messages back at the given insertion event. We
// purposely use the local user Alice here as remoteCharlie isn't even
// in the room at this point in time and even if they were, the purpose
// of this test is to make sure the remote-join will pick up the state,
// not our backfill here.
sendMarkerAndEnsureBackfilled(t, as, alice, roomID, baseInsertionEventID)

// Add some events after the marker so that remoteCharlie doesn't see the marker
createMessagesInRoom(t, alice, roomID, 3, "eventIDFiller")

// Join the room from a remote homeserver after the historical messages were sent
remoteCharlie.JoinRoom(t, roomID, []string{"hs1"})

// From the remote user, make a /context request for eventIDAfter to get
// pagination token before the marker event
contextRes := remoteCharlie.MustDoFunc(t, "GET", []string{"_matrix", "client", "r0", "rooms", roomID, "context", eventIDAfter}, client.WithContentType("application/json"), client.WithQueries(url.Values{
"limit": []string{"0"},
}))
contextResResBody := client.ParseJSON(t, contextRes)
paginationTokenBeforeMarker := client.GetJSONFieldStr(t, contextResResBody, "end")

// Start the /messages request from that pagination token which
// jumps/skips over the marker event in the timeline. This is the key
// part of the test. We want to make sure that new marker state can be
// injested and processed to reveal the imported history after a
// remote-join without paginating and backfilling over the spot in the
// timeline with the marker event.
//
// We don't want to use `validateBatchSendRes(t, remoteCharlie, roomID,
// batchSendRes, false)` here because it tests against the full message
// response and we need to skip past the marker in the timeline.
paginateUntilMessageCheckOff(t, remoteCharlie, roomID, paginationTokenBeforeMarker, historicalEventIDs, []string{})
})

t.Run("Historical messages show up for remote federated homeserver even when the homeserver is missing the part of the timeline where multiple marker events were sent and it paginates before they occured", func(t *testing.T) {
t.Parallel()

roomID := as.CreateRoom(t, createPublicRoomOpts)
alice.JoinRoom(t, roomID, nil)

// Anything above 1 here should be sufficient to test whether we can
// follow the state and previous state all the way up to injest all of
// the marker events along the way
numBatches := 2

eventIDsBefore := createMessagesInRoom(t, alice, roomID, numBatches, "eventIDsBefore")
timeAfterEventBefore := time.Now()

eventIDsAfter := createMessagesInRoom(t, alice, roomID, 3, "eventIDsAfter")
eventIDAfter := eventIDsAfter[0]

// Join the room from a remote homeserver before the historical messages were sent
remoteCharlie.JoinRoom(t, roomID, []string{"hs1"})

// Make sure all of the events have been backfilled for the remote user
// before we leave the room
fetchUntilMessagesResponseHas(t, remoteCharlie, roomID, func(ev gjson.Result) bool {
if ev.Get("event_id").Str == eventIDsBefore[0] {
return true
}

return false
})

// Leave before the historical messages are imported
remoteCharlie.LeaveRoom(t, roomID)

var expectedEventIDs []string
for i := 0; i < numBatches; i++ {
// Create separate disconnected batches
batchSendRes := batchSendHistoricalMessages(
t,
as,
roomID,
eventIDsBefore[i],
"",
createJoinStateEventsForBatchSendRequest([]string{virtualUserID}, timeAfterEventBefore),
createMessageEventsForBatchSendRequest([]string{virtualUserID}, timeAfterEventBefore, 2),
// Status
200,
)
batchSendResBody := client.ParseJSON(t, batchSendRes)
historicalEventIDs := client.GetJSONFieldStringArray(t, batchSendResBody, "event_ids")
baseInsertionEventID := client.GetJSONFieldStr(t, batchSendResBody, "base_insertion_event_id")

// Store the historical events we will expect to see later
expectedEventIDs = append(expectedEventIDs, historicalEventIDs...)

// Send the marker event which lets remote homeservers know there are
// some historical messages back at the given insertion event. We
// purposely use the local user Alice here as remoteCharlie isn't even
// in the room at this point in time and even if they were, the purpose
// of this test is to make sure the remote-join will pick up the state,
// not our backfill here.
sendMarkerAndEnsureBackfilled(t, as, alice, roomID, baseInsertionEventID)
}

// Add some events after the marker so that remoteCharlie doesn't see the marker
createMessagesInRoom(t, alice, roomID, 3, "eventIDFiller")

// Join the room from a remote homeserver after the historical messages were sent
remoteCharlie.JoinRoom(t, roomID, []string{"hs1"})

// From the remote user, make a /context request for eventIDAfter to get
// pagination token before the marker event
contextRes := remoteCharlie.MustDoFunc(t, "GET", []string{"_matrix", "client", "r0", "rooms", roomID, "context", eventIDAfter}, client.WithContentType("application/json"), client.WithQueries(url.Values{
"limit": []string{"0"},
}))
contextResResBody := client.ParseJSON(t, contextRes)
paginationTokenBeforeMarker := client.GetJSONFieldStr(t, contextResResBody, "end")

// Start the /messages request from that pagination token which
// jumps/skips over the marker event in the timeline. This is the key
// part of the test. We want to make sure that new marker state can be
// injested and processed to reveal the imported history after a
// remote-join without paginating and backfilling over the spot in the
// timeline with the marker event.
//
// We don't want to use `validateBatchSendRes(t, remoteCharlie, roomID,
// batchSendRes, false)` here because it tests against the full message
// response and we need to skip past the marker in the timeline.
paginateUntilMessageCheckOff(t, remoteCharlie, roomID, paginationTokenBeforeMarker, expectedEventIDs, []string{})
})
})

t.Run("Existing room versions", func(t *testing.T) {
Expand Down Expand Up @@ -1092,6 +1261,103 @@ func fetchUntilMessagesResponseHas(t *testing.T, c *client.CSAPI, roomID string,
}
}

// Paginate the /messages endpoint until we find all of the expectedEventIds
// (order does not matter). If any event in denyListEventIDs is found, an error
// will be thrown.
func paginateUntilMessageCheckOff(t *testing.T, c *client.CSAPI, roomID string, fromPaginationToken string, expectedEventIDs []string, denyListEventIDs []string) {
Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This function is originally from #214 but saw that we could use it here before that merges

t.Helper()
start := time.Now()

workingExpectedEventIDMap := make(map[string]string)
for _, expectedEventID := range expectedEventIDs {
workingExpectedEventIDMap[expectedEventID] = expectedEventID
}

denyEventIDMap := make(map[string]string)
for _, denyEventID := range denyListEventIDs {
denyEventIDMap[denyEventID] = denyEventID
}

var actualEventIDList []string
callCounter := 0
messageResEnd := fromPaginationToken
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",
callCounter,
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"},
"limit": []string{"100"},
"from": []string{messageResEnd},
}))
callCounter++
messsageResBody := client.ParseJSON(t, messagesRes)
messageResEnd = client.GetJSONFieldStr(t, messsageResBody, "end")
// Since the original body can only be read once, create a new one from the body bytes we just read
messagesRes.Body = ioutil.NopCloser(bytes.NewBuffer(messsageResBody))

foundEventInMessageResponse := false
must.MatchResponse(t, messagesRes, match.HTTPResponse{
JSON: []match.JSON{
match.JSONArrayEach("chunk", func(ev gjson.Result) error {
foundEventInMessageResponse = true
eventID := ev.Get("event_id").Str
actualEventIDList = append(actualEventIDList, eventID)

if _, keyExists := denyEventIDMap[eventID]; keyExists {
return fmt.Errorf(
"paginateUntilMessageCheckOff found unexpected message=%s in deny list while paginating. %s",
eventID,
generateErrorMesssageInfo(),
)
}

if _, keyExists := workingExpectedEventIDMap[eventID]; keyExists {
delete(workingExpectedEventIDMap, eventID)
}

return nil
}),
},
})

if !foundEventInMessageResponse {
t.Fatalf(
"paginateUntilMessageCheckOff reached the end of the messages without finding all expected events. %s",
generateErrorMesssageInfo(),
)
}

// We were able to find all of the expected events!
if len(workingExpectedEventIDMap) == 0 {
return
}
}
}

func historicalEventFilter(r gjson.Result) bool {
// This includes messages, insertion, batch, and marker events because we
// include the historical field on all of them.
Expand Down Expand Up @@ -1170,8 +1436,7 @@ func sendMarkerAndEnsureBackfilled(t *testing.T, as *client.CSAPI, c *client.CSA
},
}
// We can't use as.SendEventSynced(...) because application services can't use the /sync API
txnId := getTxnID("sendMarkerAndEnsureBackfilled-txn")
markerSendRes := as.MustDoFunc(t, "PUT", []string{"_matrix", "client", "r0", "rooms", roomID, "send", markerEvent.Type, txnId}, client.WithJSONBody(t, markerEvent.Content))
markerSendRes := as.MustDoFunc(t, "PUT", []string{"_matrix", "client", "r0", "rooms", roomID, "state", markerEvent.Type}, client.WithJSONBody(t, markerEvent.Content))
markerSendBody := client.ParseJSON(t, markerSendRes)
markerEventID = client.GetJSONFieldStr(t, markerSendBody, "event_id")

Expand Down