Merge branch 'main' of github.com:matrix-org/dendrite into s7evink/fts

clientapi/auth/user_interactive.go

@@ -102,7 +102,6 @@ type userInteractiveFlow struct {
 // the user already has a valid access token, but we want to double-check
 // that it isn't stolen by re-authenticating them.
 type UserInteractive struct {
-	Completed []string
 	Flows []userInteractiveFlow
 	// Map of login type to implementation
 	Types map[string]Type
@@ -116,7 +115,6 @@ func NewUserInteractive(userAccountAPI api.UserLoginAPI, cfg *config.ClientAPI)
 		Config:               cfg,
 	}
 	return &UserInteractive{
-		Completed: []string{},
 		Flows: []userInteractiveFlow{
 			{
 				Stages: []string{typePassword.Name()},
@@ -140,7 +138,6 @@ func (u *UserInteractive) IsSingleStageFlow(authType string) bool {
 
 func (u *UserInteractive) AddCompletedStage(sessionID, authType string) {
 	// TODO: Handle multi-stage flows
-	u.Completed = append(u.Completed, authType)
 	delete(u.Sessions, sessionID)
 }
 
@@ -157,7 +154,7 @@ func (u *UserInteractive) Challenge(sessionID string) *util.JSONResponse {
 	return &util.JSONResponse{
 		Code: 401,
 		JSON: Challenge{
-			Completed: u.Completed,
+			Completed: u.Sessions[sessionID],
 			Flows:     u.Flows,
 			Session:   sessionID,
 			Params:    make(map[string]interface{}),

clientapi/auth/user_interactive_test.go

@@ -187,3 +187,38 @@ func TestUserInteractivePasswordBadLogin(t *testing.T) {
 		}
 	}
 }
+
+func TestUserInteractive_AddCompletedStage(t *testing.T) {
+	tests := []struct {
+		name      string
+		sessionID string
+	}{
+		{
+			name:      "first user",
+			sessionID: util.RandomString(8),
+		},
+		{
+			name:      "second user",
+			sessionID: util.RandomString(8),
+		},
+		{
+			name:      "third user",
+			sessionID: util.RandomString(8),
+		},
+	}
+	u := setup()
+	ctx := context.Background()
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, resp := u.Verify(ctx, []byte("{}"), nil)
+			challenge, ok := resp.JSON.(Challenge)
+			if !ok {
+				t.Fatalf("expected a Challenge, got %T", resp.JSON)
+			}
+			if len(challenge.Completed) > 0 {
+				t.Fatalf("expected 0 completed stages, got %d", len(challenge.Completed))
+			}
+			u.AddCompletedStage(tt.sessionID, "")
+		})
+	}
+}

syncapi/sync/requestpool.go

@@ -251,8 +251,12 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *userapi.
 	waitingSyncRequests.Inc()
 	defer waitingSyncRequests.Dec()
 
+	// loop until we get some data
+	for {
+		startTime := time.Now()
 		currentPos := rp.Notifier.CurrentPosition()
 
+		// if the since token matches the current positions, wait via the notifier
 		if !rp.shouldReturnImmediately(syncReq, currentPos) {
 			timer := time.NewTimer(syncReq.Timeout) // case of timeout=0 is handled above
 			defer timer.Stop()
@@ -365,6 +369,27 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *userapi.
 					syncReq.Since.PresencePosition, currentPos.PresencePosition,
 				),
 			}
+			// it's possible for there to be no updates for this user even though since < current pos,
+			// e.g busy servers with a quiet user. In this scenario, we don't want to return a no-op
+			// response immediately, so let's try this again but pretend they bumped their since token.
+			// If the incremental sync was processed very quickly then we expect the next loop to block
+			// with a notifier, but if things are slow it's entirely possible that currentPos is no
+			// longer the current position so we will hit this code path again. We need to do this and
+			// not return a no-op response because:
+			// - It's an inefficient use of bandwidth.
+			// - Some sytests which test 'waking up' sync rely on some sync requests to block, which
+			//   they weren't always doing, resulting in flakey tests.
+			if !syncReq.Response.HasUpdates() {
+				syncReq.Since = currentPos
+				// do not loop again if the ?timeout= is 0 as that means "return immediately"
+				if syncReq.Timeout > 0 {
+					syncReq.Timeout = syncReq.Timeout - time.Since(startTime)
+					if syncReq.Timeout < 0 {
+						syncReq.Timeout = 0
+					}
+					continue
+				}
+			}
 		}
 
 		return util.JSONResponse{
@@ -372,6 +397,7 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *userapi.
 			JSON: syncReq.Response,
 		}
 	}
+}
 
 func (rp *RequestPool) OnIncomingKeyChangeRequest(req *http.Request, device *userapi.Device) util.JSONResponse {
 	from := req.URL.Query().Get("from")

syncapi/types/types.go

@@ -350,6 +350,19 @@ type Response struct {
 	DeviceListsOTKCount map[string]int `json:"device_one_time_keys_count,omitempty"`
 }
 
+func (r *Response) HasUpdates() bool {
+	// purposefully exclude DeviceListsOTKCount as we always include them
+	return (len(r.AccountData.Events) > 0 ||
+		len(r.Presence.Events) > 0 ||
+		len(r.Rooms.Invite) > 0 ||
+		len(r.Rooms.Join) > 0 ||
+		len(r.Rooms.Leave) > 0 ||
+		len(r.Rooms.Peek) > 0 ||
+		len(r.ToDevice.Events) > 0 ||
+		len(r.DeviceLists.Changed) > 0 ||
+		len(r.DeviceLists.Left) > 0)
+}
+
 // NewResponse creates an empty response with initialised maps.
 func NewResponse() *Response {
 	res := Response{}