Reproduce matrix-org/synapse#9768

internal/client/client.go

@@ -133,22 +133,28 @@ func (c *CSAPI) SendEventSynced(t *testing.T, roomID string, e b.Event) string {
 // - we see an event in the room for which the `check` function returns True
 // If the `check` function fails the test, the failing event will be automatically logged.
 // Will time out after CSAPI.SyncUntilTimeout.
-func (c *CSAPI) SyncUntilTimelineHas(t *testing.T, roomID string, check func(gjson.Result) bool) {
+//
+// Returns the `next_batch` token from the last /sync response. This can be passed as
+// `since` to sync from this point forward only.
+func (c *CSAPI) SyncUntilTimelineHas(t *testing.T, roomID string, check func(gjson.Result) bool) string {
 	t.Helper()
-	c.SyncUntil(t, "", "", "rooms.join."+GjsonEscape(roomID)+".timeline.events", check)
+	return c.SyncUntil(t, "", "", "rooms.join."+GjsonEscape(roomID)+".timeline.events", check)
 }
 
 // SyncUntilInvitedTo is a wrapper around SyncUntil.
 // It blocks and continually calls `/sync` until we've been invited to the given room.
 // Will time out after CSAPI.SyncUntilTimeout.
-func (c *CSAPI) SyncUntilInvitedTo(t *testing.T, roomID string) {
+//
+// Returns the `next_batch` token from the last /sync response. This can be passed as
+// `since` to sync from this point forward only.
+func (c *CSAPI) SyncUntilInvitedTo(t *testing.T, roomID string) string {
 	t.Helper()
 	check := func(event gjson.Result) bool {
 		return event.Get("type").Str == "m.room.member" &&
 			event.Get("content.membership").Str == "invite" &&
 			event.Get("state_key").Str == c.UserID
 	}
-	c.SyncUntil(t, "", "", "rooms.invite."+GjsonEscape(roomID)+".invite_state.events", check)
+	return c.SyncUntil(t, "", "", "rooms.invite."+GjsonEscape(roomID)+".invite_state.events", check)
 }
 
 // SyncUntil blocks and continually calls /sync until
@@ -157,7 +163,10 @@ func (c *CSAPI) SyncUntilInvitedTo(t *testing.T, roomID string) {
 // - some element in that array makes the `check` function return true.
 // If the `check` function fails the test, the failing event will be automatically logged.
 // Will time out after CSAPI.SyncUntilTimeout.
-func (c *CSAPI) SyncUntil(t *testing.T, since, filter, key string, check func(gjson.Result) bool) {
+//
+// Returns the `next_batch` token from the last /sync response. This can be passed as
+// `since` to sync from this point forward only.
+func (c *CSAPI) SyncUntil(t *testing.T, since, filter, key string, check func(gjson.Result) bool) string {
 	t.Helper()
 	start := time.Now()
 	checkCounter := 0
@@ -199,7 +208,7 @@ func (c *CSAPI) SyncUntil(t *testing.T, since, filter, key string, check func(gj
 			for i, ev := range events {
 				lastEvent = &events[i]
 				if check(ev) {
-					return
+					return since
 				}
 				wasFailed = t.Failed()
 				checkCounter++

tests/csapi/direct_message_create_test.go

@@ -0,0 +1,74 @@
+package csapi_tests
+
+import (
+	"testing"
+
+	"github.com/tidwall/gjson"
+
+	"github.com/matrix-org/complement/internal/b"
+	"github.com/matrix-org/complement/internal/client"
+)
+
+// Synapse has a long-standing bug
+//     https://github.com/matrix-org/synapse/issues/9768
+// where membership events are duplicated in the timeline
+// returned by a call to `/sync` with a `since` parameter.
+// (AFAICS the spec doesn't mandate that there are no such
+// duplicates, but pretty much everyone expects things to
+// work that way.)
+//
+// This test reproduces the duplicated membership event for
+// Bob after Alice invites him to a 1-1 room.
+func TestMembershipNotDuplicatedWhenJoiningDirectMessage(t *testing.T) {
+	deployment := Deploy(t, b.BlueprintAlice)
+	defer deployment.Destroy(t)
+	alice := deployment.Client(t, "hs1", "@alice:hs1")
+	bob := deployment.RegisterUser(t, "hs1", "bob", "complement_meets_min_password_req_bob")
+
+	// Alice invites Bob to a room for direct messaging.
+	roomID := alice.CreateRoom(t, map[string]interface{}{
+		"invite":    []string{bob.UserID},
+		"is_direct": true,
+	})
+
+	// Bob receives the invite and joins. Extract the
+	// `next_batch` from the response for use later.
+	nextBatch := bob.SyncUntilInvitedTo(t, roomID)
+	bob.JoinRoom(t, roomID, []string{"hs1"})
+
+	// To reproduce the bug, we ought to sync until we see
+	// a duplicate membership event. But actually seeing
+	// that event should fail the test, so we shouldn't
+	// expect it to happen. To account for both cases,
+	// send a dummy sentinel event after we've joined.
+	SENTINEL_EVENT_TYPE := "com.example.dummy"
+	bob.SendEventSynced(t, roomID, b.Event{
+		Type:    SENTINEL_EVENT_TYPE,
+		Sender:  bob.UserID,
+		Content: map[string]interface{}{},
+	})
+
+	// Replay a sync from just before we joined until we see
+	// the sentinel event. Count the number of Join events we
+	// see.
+	bobJoinEvents := 0
+	bob.SyncUntil(
+		t,
+		nextBatch,
+		"",
+		"rooms.join."+client.GjsonEscape(roomID)+".timeline.events",
+		func(ev gjson.Result) bool {
+			if ev.Get("type").Str == "m.room.member" &&
+				ev.Get("state_key").Str == bob.UserID &&
+				ev.Get("content.membership").Str == "join" {
+				bobJoinEvents += 1
+			}
+
+			return ev.Get("type").Str == SENTINEL_EVENT_TYPE
+		},
+	)
+
+	if bobJoinEvents != 1 {
+		t.Fatalf("Saw %d join events for Bob; expected 1", bobJoinEvents)
+	}
+}