Inject state blocks from incremental polls into the room timeline

state/accumulator.go

@@ -178,17 +178,21 @@ func (a *Accumulator) Initialise(roomID string, state []json.RawMessage) (Initia
 			// that the poller can log a warning.
 			logger.Debug().Str("room_id", roomID).Int64("snapshot_id", snapshotID).Msg("Accumulator.Initialise called with incremental state but current snapshot already exists.")
 			eventIDs := make([]string, len(state))
+			eventIDToRawEvent := make(map[string]json.RawMessage, len(state))
 			for i := range state {
-				eventIDs[i] = gjson.ParseBytes(state[i]).Get("event_id").Str
+				eventID := gjson.ParseBytes(state[i]).Get("event_id")
+				if !eventID.Exists() || eventID.Type != gjson.String {
+					return fmt.Errorf("Event %d lacks an event ID", i)
+				}
+				eventIDToRawEvent[eventID.Str] = state[i]
+				eventIDs[i] = eventID.Str
 			}
 			unknownEventIDs, err := a.eventsTable.SelectUnknownEventIDs(txn, eventIDs)
 			if err != nil {
 				return fmt.Errorf("error determing which event IDs are unknown: %s", err)
 			}
-			for i := range state {
-				if _, unknown := unknownEventIDs[eventIDs[i]]; unknown {
-					res.PrependTimelineEvents = append(res.PrependTimelineEvents, state[i])
-				}
+			for unknownEventID := range unknownEventIDs {
+				res.PrependTimelineEvents = append(res.PrependTimelineEvents, eventIDToRawEvent[unknownEventID])
 			}
 			return nil
 		}

state/event_table.go

@@ -260,6 +260,7 @@ func (t *EventTable) SelectStrippedEventsByIDs(txn *sqlx.Tx, verifyAll bool, ids
 }
 
 // SelectUnknownEventIDs accepts a list of event IDs and returns the subset of those which are not known to the DB.
+// It MUST be called within a transaction, or else will panic.
 func (t *EventTable) SelectUnknownEventIDs(txn *sqlx.Tx, maybeUnknownEventIDs []string) (map[string]struct{}, error) {
 	// Note: in practice, the order of rows returned matches the order of rows of
 	// array entries. But I don't think that's guaranteed. Return an (unordered) set
@@ -271,13 +272,7 @@ func (t *EventTable) SelectUnknownEventIDs(txn *sqlx.Tx, maybeUnknownEventIDs []
 	WHERE event_nid IS NULL;`
 
 	var unknownEventIDs []string
-	var err error
-	if txn != nil {
-		err = txn.Select(&unknownEventIDs, queryStr, pq.StringArray(maybeUnknownEventIDs))
-	} else {
-		err = t.db.Select(&unknownEventIDs, queryStr, pq.StringArray(maybeUnknownEventIDs))
-	}
-	if err != nil {
+	if err := txn.Select(&unknownEventIDs, queryStr, pq.StringArray(maybeUnknownEventIDs)); err != nil {
 		return nil, err
 	}
 	unknownMap := make(map[string]struct{}, len(unknownEventIDs))

state/event_table_test.go

@@ -870,3 +870,78 @@ func TestRemoveUnsignedTXNID(t *testing.T) {
 		}
 	}
 }
+
+func TestEventTableSelectUnknownEventIDs(t *testing.T) {
+	db, close := connectToDB(t)
+	defer close()
+	txn, err := db.Beginx()
+	if err != nil {
+		t.Fatalf("failed to start txn: %s", err)
+	}
+	defer txn.Rollback()
+	const roomID = "!1:localhost"
+
+	// Note: there shouldn't be any other events with these IDs inserted before this
+	// transaction. $A and $B seem to be inserted and commit in TestEventTablePrevBatch.
+	const eventID1 = "$A-SelectUnknownEventIDs"
+	const eventID2 = "$B-SelectUnknownEventIDs"
+
+	knownEvents := []Event{
+		{
+			Type:     "m.room.create",
+			StateKey: "",
+			IsState:  true,
+			ID:       eventID1,
+			RoomID:   roomID,
+		},
+		{
+			Type:     "m.room.name",
+			StateKey: "",
+			IsState:  true,
+			ID:       eventID2,
+			RoomID:   roomID,
+		},
+	}
+	table := NewEventTable(db)
+
+	// Check the event IDs haven't been added by another test.
+	gotEvents, err := table.SelectByIDs(txn, true, []string{eventID1, eventID2})
+	if len(gotEvents) > 0 {
+		t.Fatalf("Event IDs already in use---commited by another test?")
+	}
+
+	// Insert the events
+	_, err = table.Insert(txn, knownEvents, false)
+	if err != nil {
+		t.Fatalf("failed to insert event: %s", err)
+	}
+
+	gotEvents, err = table.SelectByIDs(txn, true, []string{eventID1, eventID2})
+	if err != nil {
+		t.Fatalf("failed to select events: %s", err)
+	}
+	if (gotEvents[0].ID == eventID1 && gotEvents[1].ID == eventID2) || (gotEvents[0].ID == eventID2 && gotEvents[1].ID == eventID1) {
+		t.Logf("Got expected event IDs after insert. NIDS: %s=%d, %s=%d", gotEvents[0].ID, gotEvents[0].NID, gotEvents[1].ID, gotEvents[1].NID)
+	} else {
+		t.Fatalf("Event ID mismatch: expected $A-SelectUnknownEventIDs and $B-SelectUnknownEventIDs, got %v", gotEvents)
+	}
+
+	// Someone else tells us the state of the room is {A, C}. Query which of those
+	// event IDs are unknown.
+	const unknownEventID = "$C-SelectUnknownEventIDs"
+	stateBlockIDs := []string{eventID1, unknownEventID}
+	unknownIDs, err := table.SelectUnknownEventIDs(txn, stateBlockIDs)
+	t.Logf("unknownIDs=%v", unknownIDs)
+	if err != nil {
+		t.Fatalf("failed to select unknown state events: %s", err)
+	}
+
+	// Only event C should be flagged as unknown.
+	if len(unknownIDs) != 1 {
+		t.Fatalf("Expected 1 unknown id, got %v", unknownIDs)
+	}
+	_, ok := unknownIDs[unknownEventID]
+	if !ok {
+		t.Fatalf("Expected $C-SelectUnknownEventIDs to be unknown to the DB, but it wasn't")
+	}
+}

tests-e2e/gappy_state_test.go

@@ -4,7 +4,6 @@ import (
 	"fmt"
 	"github.com/matrix-org/sliding-sync/sync3"
 	"github.com/matrix-org/sliding-sync/testutils/m"
-	"github.com/tidwall/gjson"
 	"testing"
 )
 
@@ -46,16 +45,7 @@ func TestGappyState(t *testing.T) {
 		m.MatchRoomSubscription(
 			roomID,
 			m.MatchRoomName(firstRoomName),
-			func(r sync3.Room) error {
-				if len(r.Timeline) == 0 {
-					return fmt.Errorf("no/empty timeline in response")
-				}
-				lastReceivedEventID := gjson.ParseBytes(r.Timeline[len(r.Timeline)-1]).Get("event_id").Str
-				if lastReceivedEventID == firstMessageID {
-					return nil
-				}
-				return fmt.Errorf("expected end of timeline to be %s but got %s", firstMessageID, lastReceivedEventID)
-			},
+			MatchRoomTimelineMostRecent(1, []Event{{ID: firstMessageID}}),
 		),
 	)
 
@@ -102,16 +92,7 @@ func TestGappyState(t *testing.T) {
 		m.MatchRoomSubscription(
 			roomID,
 			m.MatchRoomName("potato"),
-			func(r sync3.Room) error {
-				if len(r.Timeline) == 0 {
-					return fmt.Errorf("no timeline in response, expected at least one event")
-				}
-				lastReceivedEventID := gjson.ParseBytes(r.Timeline[len(r.Timeline)-1]).Get("event_id").Str
-				if lastReceivedEventID != latestMessageID {
-					return fmt.Errorf("last message in response is %s, expected %s", lastReceivedEventID, latestMessageID)
-				}
-				return nil
-			},
+			MatchRoomTimelineMostRecent(1, []Event{{ID: latestMessageID}}),
 		),
 	)
 }

tests-e2e/main_test.go

@@ -79,10 +79,22 @@ func eventsEqual(wantList []Event, gotList []json.RawMessage) error {
 	return nil
 }
 
+// MatchRoomTimelineMostRecent builds a matcher which checks that the last `n` elements
+// of `events` are the same as the last n elements of the room timeline. If either list
+// contains fewer than `n` events, the match fails.
+// Events are tested for equality using `eventsEqual`.
 func MatchRoomTimelineMostRecent(n int, events []Event) m.RoomMatcher {
-	subset := events[len(events)-n:]
 	return func(r sync3.Room) error {
-		return MatchRoomTimeline(subset)(r)
+		if len(events) < n {
+			return fmt.Errorf("list of wanted events has %d events, expected at least %d", len(events), n)
+		}
+		wantList := events[len(events)-n:]
+		if len(r.Timeline) < n {
+			return fmt.Errorf("timeline has %d events, expected at least %d", len(r.Timeline), n)
+		}
+
+		gotList := r.Timeline[len(r.Timeline)-n:]
+		return eventsEqual(wantList, gotList)
 	}
 }
 

tests-integration/poller_test.go

@@ -113,33 +113,38 @@ func TestSecondPollerFiltersToDevice(t *testing.T) {
 	m.MatchResponse(t, res, m.MatchToDeviceMessages([]json.RawMessage{wantMsg}))
 }
 
-// TODO test description
+// Test that the poller makes a best-effort attempt to integrate state seen in a
+// v2 sync state block. Our strategy for doing so is to prepend any unknown state events
+// to the start of the v2 sync response's timeline, which should then be visible to
+// sync v3 clients as ordinary state events in the room timeline.
 func TestPollerHandlesUnknownStateEventsOnIncrementalSync(t *testing.T) {
 	pqString := testutils.PrepareDBConnectionString()
 	v2 := runTestV2Server(t)
 	v3 := runTestServer(t, v2, pqString)
 	defer v2.close()
 	defer v3.close()
-	deviceAToken := "DEVICE_A_TOKEN"
-	v2.addAccount(alice, deviceAToken)
+	v2.addAccount(alice, aliceToken)
 	const roomID = "!unimportant"
-	v2.queueResponse(deviceAToken, sync2.SyncResponse{
+	v2.queueResponse(aliceToken, sync2.SyncResponse{
 		Rooms: sync2.SyncRoomsResponse{
 			Join: v2JoinTimeline(roomEvents{
 				roomID: roomID,
 				events: createRoomState(t, alice, time.Now()),
 			}),
 		},
 	})
-	res := v3.mustDoV3Request(t, deviceAToken, sync3.Request{
+	res := v3.mustDoV3Request(t, aliceToken, sync3.Request{
 		Lists: map[string]sync3.RequestList{
 			"a": {
 				Ranges: [][2]int64{{0, 20}},
+				RoomSubscription: sync3.RoomSubscription{
+					TimelineLimit: 10,
+				},
 			},
 		},
 	})
 
-	t.Log("The poller receives a gappy incremental sync response with a state block")
+	t.Log("The poller receives a gappy incremental sync response with a state block. The power levels and room name have changed.")
 	nameEvent := testutils.NewStateEvent(
 		t,
 		"m.room.name",
@@ -158,7 +163,7 @@ func TestPollerHandlesUnknownStateEventsOnIncrementalSync(t *testing.T) {
 		},
 	)
 	messageEvent := testutils.NewMessageEvent(t, alice, "hello")
-	v2.queueResponse(deviceAToken, sync2.SyncResponse{
+	v2.queueResponse(aliceToken, sync2.SyncResponse{
 		Rooms: sync2.SyncRoomsResponse{
 			Join: map[string]sync2.SyncV2JoinResponse{
 				roomID: {
@@ -175,7 +180,7 @@ func TestPollerHandlesUnknownStateEventsOnIncrementalSync(t *testing.T) {
 		},
 	})
 
-	res = v3.mustDoV3RequestWithPos(t, deviceAToken, res.Pos, sync3.Request{})
+	res = v3.mustDoV3RequestWithPos(t, aliceToken, res.Pos, sync3.Request{})
 	m.MatchResponse(
 		t,
 		res,
@@ -186,3 +191,104 @@ func TestPollerHandlesUnknownStateEventsOnIncrementalSync(t *testing.T) {
 		),
 	)
 }
+
+// Similar to TestPollerHandlesUnknownStateEventsOnIncrementalSync. Here we are testing
+// that if Alice's poller sees Bob leave in a state block, the events seen in that
+// timeline are not visible to Bob.
+func TestPollerUpdatesRoomMemberTrackerOnGappySyncStateBlock(t *testing.T) {
+	pqString := testutils.PrepareDBConnectionString()
+	v2 := runTestV2Server(t)
+	v3 := runTestServer(t, v2, pqString)
+	defer v2.close()
+	defer v3.close()
+	v2.addAccount(alice, aliceToken)
+	v2.addAccount(bob, bobToken)
+	const roomID = "!unimportant"
+
+	t.Log("Alice's poller does an initial sync. It sees that Alice and Bob share a room.")
+	initialTimeline := createRoomState(t, alice, time.Now())
+	bobJoin := testutils.NewStateEvent(
+		t,
+		"m.room.member",
+		bob,
+		bob,
+		map[string]interface{}{"membership": "join"},
+	)
+	initialJoinBlock := v2JoinTimeline(roomEvents{
+		roomID: roomID,
+		events: append(initialTimeline, bobJoin),
+	})
+	v2.queueResponse(aliceToken, sync2.SyncResponse{
+		Rooms: sync2.SyncRoomsResponse{Join: initialJoinBlock},
+	})
+
+	t.Log("Alice makes an initial sliding sync request.")
+	aliceRes := v3.mustDoV3Request(t, aliceToken, sync3.Request{
+		Lists: map[string]sync3.RequestList{
+			"a": {
+				Ranges: [][2]int64{{0, 20}},
+				RoomSubscription: sync3.RoomSubscription{
+					TimelineLimit: 10,
+				},
+			},
+		},
+	})
+
+	t.Log("Alice's poller receives a gappy incremental sync response. Bob has left in the gap. The timeline includes a message from Alice.")
+	bobLeave := testutils.NewStateEvent(
+		t,
+		"m.room.member",
+		bob,
+		bob,
+		map[string]interface{}{"membership": "leave"},
+	)
+	aliceMessage := testutils.NewMessageEvent(t, alice, "hello")
+	v2.queueResponse(aliceToken, sync2.SyncResponse{
+		Rooms: sync2.SyncRoomsResponse{
+			Join: map[string]sync2.SyncV2JoinResponse{
+				roomID: {
+					State: sync2.EventsResponse{
+						Events: []json.RawMessage{bobLeave},
+					},
+					Timeline: sync2.TimelineResponse{
+						Events:    []json.RawMessage{aliceMessage},
+						Limited:   true,
+						PrevBatch: "batchymcbatchface",
+					},
+				},
+			},
+		},
+	})
+
+	t.Log("Alice makes an incremental sliding sync request.")
+	aliceRes = v3.mustDoV3RequestWithPos(t, aliceToken, aliceRes.Pos, sync3.Request{})
+
+	t.Log("She should see Bob's leave event and her message at the end of the room timeline.")
+	m.MatchResponse(
+		t,
+		aliceRes,
+		m.MatchRoomSubscription(
+			roomID,
+			m.MatchRoomTimelineMostRecent(2, []json.RawMessage{bobLeave, aliceMessage}),
+		),
+	)
+
+	t.Log("Bob makes an initial sliding sync request.")
+	bobRes := v3.mustDoV3Request(t, bobToken, sync3.Request{
+		Lists: map[string]sync3.RequestList{
+			"a": {
+				Ranges: [][2]int64{{0, 20}},
+				RoomSubscription: sync3.RoomSubscription{
+					TimelineLimit: 10,
+				},
+			},
+		},
+	})
+	t.Log("He should not see himself in the room.")
+	m.MatchResponse(
+		t,
+		bobRes,
+		m.MatchList("a", m.MatchV3Count(0)),
+	)
+
+}