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

.github/workflows/dendrite.yml

@@ -109,6 +109,11 @@ jobs:
         uses: actions/setup-go@v3
         with:
           go-version: ${{ matrix.go }}
+      - name: Set up gotestfmt
+        uses: gotesttools/gotestfmt-action@v2
+        with:
+          # Optional: pass GITHUB_TOKEN to avoid rate limiting.
+          token: ${{ secrets.GITHUB_TOKEN }}
       - uses: actions/cache@v3
         with:
           path: |
@@ -117,7 +122,7 @@ jobs:
           key: ${{ runner.os }}-go${{ matrix.go }}-test-${{ hashFiles('**/go.sum') }}
           restore-keys: |
             ${{ runner.os }}-go${{ matrix.go }}-test-
-      - run: go test ./...
+      - run: go test -json -v ./... 2>&1 | gotestfmt
         env:
           POSTGRES_HOST: localhost
           POSTGRES_USER: postgres

CHANGES.md

@@ -1,5 +1,23 @@
 # Changelog
 
+## Dendrite 0.10.4 (2022-10-21)
+
+### Features
+
+* Various tables belonging to the user API will be renamed so that they are namespaced with the `userapi_` prefix
+  * Note that, after upgrading to this version, you should not revert to an older version of Dendrite as the database changes **will not** be reverted automatically
+* The backoff and retry behaviour in the federation API has been refactored and improved
+
+### Fixes
+
+* Private read receipt support is now advertised in the client `/versions` endpoint
+* Private read receipts will now clear notification counts properly
+* A bug where a false `leave` membership transition was inserted into the timeline after accepting an invite has been fixed
+* Some panics caused by concurrent map writes in the key server have been fixed
+* The sync API now calculates membership transitions from state deltas more accurately
+* Transaction IDs are now scoped to endpoints, which should fix some bugs where transaction ID reuse could cause nonsensical cached responses from some endpoints
+* The length of the `type`, `sender`, `state_key` and `room_id` fields in events are now verified by number of bytes rather than codepoints after a spec clarification, reverting a change made in Dendrite 0.9.6
+
 ## Dendrite 0.10.3 (2022-10-14)
 
 ### Features

clientapi/routing/joined_rooms.go

@@ -0,0 +1,52 @@
+// Copyright 2022 The Matrix.org Foundation C.I.C.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package routing
+
+import (
+	"net/http"
+
+	"github.com/matrix-org/util"
+
+	"github.com/matrix-org/dendrite/clientapi/jsonerror"
+	"github.com/matrix-org/dendrite/roomserver/api"
+	userapi "github.com/matrix-org/dendrite/userapi/api"
+)
+
+type getJoinedRoomsResponse struct {
+	JoinedRooms []string `json:"joined_rooms"`
+}
+
+func GetJoinedRooms(
+	req *http.Request,
+	device *userapi.Device,
+	rsAPI api.ClientRoomserverAPI,
+) util.JSONResponse {
+	var res api.QueryRoomsForUserResponse
+	err := rsAPI.QueryRoomsForUser(req.Context(), &api.QueryRoomsForUserRequest{
+		UserID:         device.UserID,
+		WantMembership: "join",
+	}, &res)
+	if err != nil {
+		util.GetLogger(req.Context()).WithError(err).Error("QueryRoomsForUser failed")
+		return jsonerror.InternalServerError()
+	}
+	if res.RoomIDs == nil {
+		res.RoomIDs = []string{}
+	}
+	return util.JSONResponse{
+		Code: http.StatusOK,
+		JSON: getJoinedRoomsResponse{res.RoomIDs},
+	}
+}

clientapi/routing/keys.go

@@ -99,7 +99,11 @@ func (r *queryKeysRequest) GetTimeout() time.Duration {
 	if r.Timeout == 0 {
 		return 10 * time.Second
 	}
-	return time.Duration(r.Timeout) * time.Millisecond
+	timeout := time.Duration(r.Timeout) * time.Millisecond
+	if timeout > time.Second*20 {
+		timeout = time.Second * 20
+	}
+	return timeout
 }
 
 func QueryKeys(req *http.Request, keyAPI api.ClientKeyAPI, device *userapi.Device) util.JSONResponse {

clientapi/routing/routing.go

@@ -950,26 +950,6 @@ func Setup(
 		}),
 	).Methods(http.MethodPost, http.MethodOptions)
 
-	v3mux.Handle("/rooms/{roomID}/members",
-		httputil.MakeAuthAPI("rooms_members", userAPI, func(req *http.Request, device *userapi.Device) util.JSONResponse {
-			vars, err := httputil.URLDecodeMapValues(mux.Vars(req))
-			if err != nil {
-				return util.ErrorResponse(err)
-			}
-			return GetMemberships(req, device, vars["roomID"], false, cfg, rsAPI)
-		}),
-	).Methods(http.MethodGet, http.MethodOptions)
-
-	v3mux.Handle("/rooms/{roomID}/joined_members",
-		httputil.MakeAuthAPI("rooms_members", userAPI, func(req *http.Request, device *userapi.Device) util.JSONResponse {
-			vars, err := httputil.URLDecodeMapValues(mux.Vars(req))
-			if err != nil {
-				return util.ErrorResponse(err)
-			}
-			return GetMemberships(req, device, vars["roomID"], true, cfg, rsAPI)
-		}),
-	).Methods(http.MethodGet, http.MethodOptions)
-
 	v3mux.Handle("/rooms/{roomID}/read_markers",
 		httputil.MakeAuthAPI("rooms_read_markers", userAPI, func(req *http.Request, device *userapi.Device) util.JSONResponse {
 			if r := rateLimits.Limit(req, device); r != nil {

docs/caddy/polylith/Caddyfile

@@ -74,7 +74,7 @@ matrix.example.com {
  # Change the end of each reverse_proxy line to the correct
  # address for your various services.
  @sync_api {
-  path_regexp /_matrix/client/.*?/(sync|user/.*?/filter/?.*|keys/changes|rooms/.*?/(messages|context/.*?|relations/.*?|event/.*?))$
+  path_regexp /_matrix/client/.*?/(sync|user/.*?/filter/?.*|keys/changes|rooms/.*?/(messages|.*?_?members|context/.*?|relations/.*?|event/.*?))$
  }
  reverse_proxy @sync_api sync_api:8073
 

docs/hiawatha/polylith-sample.conf

@@ -23,8 +23,10 @@ VirtualHost {
         # /_matrix/client/.*/rooms/{roomId}/relations/{eventID}
         # /_matrix/client/.*/rooms/{roomId}/relations/{eventID}/{relType}
         # /_matrix/client/.*/rooms/{roomId}/relations/{eventID}/{relType}/{eventType}
+        # /_matrix/client/.*/rooms/{roomId}/members
+        # /_matrix/client/.*/rooms/{roomId}/joined_members
         # to sync_api
-        ReverseProxy = /_matrix/client/.*?/(sync|user/.*?/filter/?.*|keys/changes|rooms/.*?/(messages|context/.*?|relations/.*?|event/.*?))$ http://localhost:8073 600
+        ReverseProxy = /_matrix/client/.*?/(sync|user/.*?/filter/?.*|keys/changes|rooms/.*?/(messages|.*?_?members|context/.*?|relations/.*?|event/.*?))$ http://localhost:8073 600
         ReverseProxy = /_matrix/client http://localhost:8071 600
         ReverseProxy = /_matrix/federation http://localhost:8072 600
         ReverseProxy = /_matrix/key http://localhost:8072 600

docs/nginx/polylith-sample.conf

@@ -33,8 +33,10 @@ server {
     # /_matrix/client/.*/rooms/{roomId}/relations/{eventID}
     # /_matrix/client/.*/rooms/{roomId}/relations/{eventID}/{relType}
     # /_matrix/client/.*/rooms/{roomId}/relations/{eventID}/{relType}/{eventType}
+    # /_matrix/client/.*/rooms/{roomId}/members
+    # /_matrix/client/.*/rooms/{roomId}/joined_members
     # to sync_api
-    location ~ /_matrix/client/.*?/(sync|user/.*?/filter/?.*|keys/changes|rooms/.*?/(messages|context/.*?|relations/.*?|event/.*?))$  {
+    location ~ /_matrix/client/.*?/(sync|user/.*?/filter/?.*|keys/changes|rooms/.*?/(messages|.*?_?members|context/.*?|relations/.*?|event/.*?))$  {
         proxy_pass http://sync_api:8073;
     }
 

internal/version.go

@@ -17,7 +17,7 @@ var build string
 const (
 	VersionMajor = 0
 	VersionMinor = 10
-	VersionPatch = 3
+	VersionPatch = 4
 	VersionTag   = "" // example: "rc1"
 )
 

keyserver/internal/internal.go

@@ -257,9 +257,6 @@ func (a *KeyInternalAPI) QueryKeys(ctx context.Context, req *api.QueryKeysReques
 	res.UserSigningKeys = make(map[string]gomatrixserverlib.CrossSigningKey)
 	res.Failures = make(map[string]interface{})
 
-	// get cross-signing keys from the database
-	a.crossSigningKeysFromDatabase(ctx, req, res)
-
 	// make a map from domain to device keys
 	domainToDeviceKeys := make(map[string]map[string][]string)
 	domainToCrossSigningKeys := make(map[string]map[string]struct{})
@@ -336,6 +333,10 @@ func (a *KeyInternalAPI) QueryKeys(ctx context.Context, req *api.QueryKeysReques
 		a.queryRemoteKeys(ctx, req.Timeout, res, domainToDeviceKeys, domainToCrossSigningKeys)
 	}
 
+	// Now that we've done the potentially expensive work of asking the federation,
+	// try filling the cross-signing keys from the database that we know about.
+	a.crossSigningKeysFromDatabase(ctx, req, res)
+
 	// Finally, append signatures that we know about
 	// TODO: This is horrible because we need to round-trip the signature from
 	// JSON, add the signatures and marshal it again, for some reason?

keyserver/storage/postgres/cross_signing_sigs_table.go

@@ -42,7 +42,7 @@ CREATE INDEX IF NOT EXISTS keyserver_cross_signing_sigs_idx ON keyserver_cross_s
 
 const selectCrossSigningSigsForTargetSQL = "" +
 	"SELECT origin_user_id, origin_key_id, signature FROM keyserver_cross_signing_sigs" +
-	" WHERE (origin_user_id = $1 OR origin_user_id = target_user_id) AND target_user_id = $2 AND target_key_id = $3"
+	" WHERE (origin_user_id = $1 OR origin_user_id = $2) AND target_user_id = $2 AND target_key_id = $3"
 
 const upsertCrossSigningSigsForTargetSQL = "" +
 	"INSERT INTO keyserver_cross_signing_sigs (origin_user_id, origin_key_id, target_user_id, target_key_id, signature)" +

keyserver/storage/sqlite3/cross_signing_sigs_table.go

@@ -42,7 +42,7 @@ CREATE INDEX IF NOT EXISTS keyserver_cross_signing_sigs_idx ON keyserver_cross_s
 
 const selectCrossSigningSigsForTargetSQL = "" +
 	"SELECT origin_user_id, origin_key_id, signature FROM keyserver_cross_signing_sigs" +
-	" WHERE (origin_user_id = $1 OR origin_user_id = target_user_id) AND target_user_id = $2 AND target_key_id = $3"
+	" WHERE (origin_user_id = $1 OR origin_user_id = $2) AND target_user_id = $3 AND target_key_id = $4"
 
 const upsertCrossSigningSigsForTargetSQL = "" +
 	"INSERT OR REPLACE INTO keyserver_cross_signing_sigs (origin_user_id, origin_key_id, target_user_id, target_key_id, signature)" +
@@ -85,7 +85,7 @@ func NewSqliteCrossSigningSigsTable(db *sql.DB) (tables.CrossSigningSigs, error)
 func (s *crossSigningSigsStatements) SelectCrossSigningSigsForTarget(
 	ctx context.Context, txn *sql.Tx, originUserID, targetUserID string, targetKeyID gomatrixserverlib.KeyID,
 ) (r types.CrossSigningSigMap, err error) {
-	rows, err := sqlutil.TxStmt(txn, s.selectCrossSigningSigsForTargetStmt).QueryContext(ctx, originUserID, targetUserID, targetKeyID)
+	rows, err := sqlutil.TxStmt(txn, s.selectCrossSigningSigsForTargetStmt).QueryContext(ctx, originUserID, targetUserID, targetUserID, targetKeyID)
 	if err != nil {
 		return nil, err
 	}

setup/config/config_global.go

@@ -170,7 +170,7 @@ type ServerNotices struct {
 	// The displayname to be used when sending notices
 	DisplayName string `yaml:"display_name"`
 	// The avatar of this user
-	AvatarURL string `yaml:"avatar"`
+	AvatarURL string `yaml:"avatar_url"`
 	// The roomname to be used when creating messages
 	RoomName string `yaml:"room_name"`
 }

syncapi/routing/memberships.go

@@ -18,22 +18,20 @@ import (
 	"encoding/json"
 	"net/http"
 
-	"github.com/matrix-org/dendrite/clientapi/jsonerror"
-	"github.com/matrix-org/dendrite/roomserver/api"
-	"github.com/matrix-org/dendrite/setup/config"
-	userapi "github.com/matrix-org/dendrite/userapi/api"
 	"github.com/matrix-org/gomatrixserverlib"
 	"github.com/matrix-org/util"
+
+	"github.com/matrix-org/dendrite/clientapi/jsonerror"
+	"github.com/matrix-org/dendrite/roomserver/api"
+	"github.com/matrix-org/dendrite/syncapi/storage"
+	"github.com/matrix-org/dendrite/syncapi/types"
+	userapi "github.com/matrix-org/dendrite/userapi/api"
 )
 
 type getMembershipResponse struct {
 	Chunk []gomatrixserverlib.ClientEvent `json:"chunk"`
 }
 
-type getJoinedRoomsResponse struct {
-	JoinedRooms []string `json:"joined_rooms"`
-}
-
 // https://matrix.org/docs/spec/client_server/r0.6.0#get-matrix-client-r0-rooms-roomid-joined-members
 type getJoinedMembersResponse struct {
 	Joined map[string]joinedMember `json:"joined"`
@@ -51,19 +49,22 @@ type databaseJoinedMember struct {
 	AvatarURL   string `json:"avatar_url"`
 }
 
-// GetMemberships implements GET /rooms/{roomId}/members
+// GetMemberships implements
+//
+//	GET /rooms/{roomId}/members
+//	GET /rooms/{roomId}/joined_members
 func GetMemberships(
-	req *http.Request, device *userapi.Device, roomID string, joinedOnly bool,
-	_ *config.ClientAPI,
-	rsAPI api.ClientRoomserverAPI,
+	req *http.Request, device *userapi.Device, roomID string,
+	syncDB storage.Database, rsAPI api.SyncRoomserverAPI,
+	joinedOnly bool, membership, notMembership *string, at string,
 ) util.JSONResponse {
-	queryReq := api.QueryMembershipsForRoomRequest{
-		JoinedOnly: joinedOnly,
-		RoomID:     roomID,
-		Sender:     device.UserID,
+	queryReq := api.QueryMembershipForUserRequest{
+		RoomID: roomID,
+		UserID: device.UserID,
 	}
-	var queryRes api.QueryMembershipsForRoomResponse
-	if err := rsAPI.QueryMembershipsForRoom(req.Context(), &queryReq, &queryRes); err != nil {
+
+	var queryRes api.QueryMembershipForUserResponse
+	if err := rsAPI.QueryMembershipForUser(req.Context(), &queryReq, &queryRes); err != nil {
 		util.GetLogger(req.Context()).WithError(err).Error("rsAPI.QueryMembershipsForRoom failed")
 		return jsonerror.InternalServerError()
 	}
@@ -75,16 +76,54 @@ func GetMemberships(
 		}
 	}
 
+	db, err := syncDB.NewDatabaseSnapshot(req.Context())
+	if err != nil {
+		return jsonerror.InternalServerError()
+	}
+
+	atToken, err := types.NewTopologyTokenFromString(at)
+	if err != nil {
+		if queryRes.HasBeenInRoom && !queryRes.IsInRoom {
+			// If you have left the room then this will be the members of the room when you left.
+			atToken, err = db.EventPositionInTopology(req.Context(), queryRes.EventID)
+		} else {
+			// If you are joined to the room then this will be the current members of the room.
+			atToken, err = db.MaxTopologicalPosition(req.Context(), roomID)
+		}
+		if err != nil {
+			util.GetLogger(req.Context()).WithError(err).Error("unable to get 'atToken'")
+			return jsonerror.InternalServerError()
+		}
+	}
+
+	eventIDs, err := db.SelectMemberships(req.Context(), roomID, atToken, membership, notMembership)
+	if err != nil {
+		util.GetLogger(req.Context()).WithError(err).Error("db.SelectMemberships failed")
+		return jsonerror.InternalServerError()
+	}
+
+	result, err := db.Events(req.Context(), eventIDs)
+	if err != nil {
+		util.GetLogger(req.Context()).WithError(err).Error("db.Events failed")
+		return jsonerror.InternalServerError()
+	}
+
 	if joinedOnly {
+		if !queryRes.IsInRoom {
+			return util.JSONResponse{
+				Code: http.StatusForbidden,
+				JSON: jsonerror.Forbidden("You aren't a member of the room and weren't previously a member of the room."),
+			}
+		}
 		var res getJoinedMembersResponse
 		res.Joined = make(map[string]joinedMember)
-		for _, ev := range queryRes.JoinEvents {
+		for _, ev := range result {
 			var content databaseJoinedMember
-			if err := json.Unmarshal(ev.Content, &content); err != nil {
+			if err := json.Unmarshal(ev.Content(), &content); err != nil {
 				util.GetLogger(req.Context()).WithError(err).Error("failed to unmarshal event content")
 				return jsonerror.InternalServerError()
 			}
-			res.Joined[ev.Sender] = joinedMember(content)
+			res.Joined[ev.Sender()] = joinedMember(content)
 		}
 		return util.JSONResponse{
 			Code: http.StatusOK,
@@ -93,29 +132,6 @@ func GetMemberships(
 	}
 	return util.JSONResponse{
 		Code: http.StatusOK,
-		JSON: getMembershipResponse{queryRes.JoinEvents},
-	}
-}
-
-func GetJoinedRooms(
-	req *http.Request,
-	device *userapi.Device,
-	rsAPI api.ClientRoomserverAPI,
-) util.JSONResponse {
-	var res api.QueryRoomsForUserResponse
-	err := rsAPI.QueryRoomsForUser(req.Context(), &api.QueryRoomsForUserRequest{
-		UserID:         device.UserID,
-		WantMembership: "join",
-	}, &res)
-	if err != nil {
-		util.GetLogger(req.Context()).WithError(err).Error("QueryRoomsForUser failed")
-		return jsonerror.InternalServerError()
-	}
-	if res.RoomIDs == nil {
-		res.RoomIDs = []string{}
-	}
-	return util.JSONResponse{
-		Code: http.StatusOK,
-		JSON: getJoinedRoomsResponse{res.RoomIDs},
+		JSON: getMembershipResponse{gomatrixserverlib.HeaderedToClientEvents(result, gomatrixserverlib.FormatSync)},
 	}
 }

syncapi/routing/messages.go

@@ -83,18 +83,18 @@ func OnIncomingMessagesRequest(
 	defer sqlutil.EndTransactionWithCheck(snapshot, &succeeded, &err)
 
 	// check if the user has already forgotten about this room
-	isForgotten, roomExists, err := checkIsRoomForgotten(req.Context(), roomID, device.UserID, rsAPI)
+	membershipResp, err := getMembershipForUser(req.Context(), roomID, device.UserID, rsAPI)
 	if err != nil {
 		return jsonerror.InternalServerError()
 	}
-	if !roomExists {
+	if !membershipResp.RoomExists {
 		return util.JSONResponse{
 			Code: http.StatusForbidden,
 			JSON: jsonerror.Forbidden("room does not exist"),
 		}
 	}
 
-	if isForgotten {
+	if membershipResp.IsRoomForgotten {
 		return util.JSONResponse{
 			Code: http.StatusForbidden,
 			JSON: jsonerror.Forbidden("user already forgot about this room"),
@@ -195,6 +195,20 @@ func OnIncomingMessagesRequest(
 		}
 	}
 
+	// If the user already left the room, grep events from before that
+	if membershipResp.Membership == gomatrixserverlib.Leave {
+		var token types.TopologyToken
+		token, err = snapshot.EventPositionInTopology(req.Context(), membershipResp.EventID)
+		if err != nil {
+			return util.JSONResponse{
+				Code: http.StatusInternalServerError,
+			}
+		}
+		if backwardOrdering {
+			from = token
+		}
+	}
+
 	mReq := messagesReq{
 		ctx:              req.Context(),
 		db:               db,
@@ -283,17 +297,16 @@ func (m *messagesResp) applyLazyLoadMembers(
 	}
 }
 
-func checkIsRoomForgotten(ctx context.Context, roomID, userID string, rsAPI api.SyncRoomserverAPI) (forgotten bool, exists bool, err error) {
+func getMembershipForUser(ctx context.Context, roomID, userID string, rsAPI api.SyncRoomserverAPI) (resp api.QueryMembershipForUserResponse, err error) {
 	req := api.QueryMembershipForUserRequest{
 		RoomID: roomID,
 		UserID: userID,
 	}
-	resp := api.QueryMembershipForUserResponse{}
 	if err := rsAPI.QueryMembershipForUser(ctx, &req, &resp); err != nil {
-		return false, false, err
+		return api.QueryMembershipForUserResponse{}, err
 	}
 
-	return resp.IsRoomForgotten, resp.RoomExists, nil
+	return resp, nil
 }
 
 // retrieveEvents retrieves events from the local database for a request on
@@ -313,7 +326,11 @@ func (r *messagesReq) retrieveEvents() (
 	}
 
 	var events []*gomatrixserverlib.HeaderedEvent
-	util.GetLogger(r.ctx).WithField("start", start).WithField("end", end).Infof("Fetched %d events locally", len(streamEvents))
+	util.GetLogger(r.ctx).WithFields(logrus.Fields{
+		"start":     r.from,
+		"end":       r.to,
+		"backwards": r.backwardOrdering,
+	}).Infof("Fetched %d events locally", len(streamEvents))
 
 	// There can be two reasons for streamEvents to be empty: either we've
 	// reached the oldest event in the room (or the most recent one, depending

syncapi/routing/routing.go

@@ -172,4 +172,37 @@ func Setup(
 			return Search(req, device, syncDB, fts, nextBatch)
 		}),
 	).Methods(http.MethodPost, http.MethodOptions)
+
+	v3mux.Handle("/rooms/{roomID}/members",
+		httputil.MakeAuthAPI("rooms_members", userAPI, func(req *http.Request, device *userapi.Device) util.JSONResponse {
+			vars, err := httputil.URLDecodeMapValues(mux.Vars(req))
+			if err != nil {
+				return util.ErrorResponse(err)
+			}
+			var membership, notMembership *string
+			if req.URL.Query().Has("membership") {
+				m := req.URL.Query().Get("membership")
+				membership = &m
+			}
+			if req.URL.Query().Has("not_membership") {
+				m := req.URL.Query().Get("not_membership")
+				notMembership = &m
+			}
+
+			at := req.URL.Query().Get("at")
+			return GetMemberships(req, device, vars["roomID"], syncDB, rsAPI, false, membership, notMembership, at)
+		}),
+	).Methods(http.MethodGet, http.MethodOptions)
+
+	v3mux.Handle("/rooms/{roomID}/joined_members",
+		httputil.MakeAuthAPI("rooms_members", userAPI, func(req *http.Request, device *userapi.Device) util.JSONResponse {
+			vars, err := httputil.URLDecodeMapValues(mux.Vars(req))
+			if err != nil {
+				return util.ErrorResponse(err)
+			}
+			at := req.URL.Query().Get("at")
+			membership := gomatrixserverlib.Join
+			return GetMemberships(req, device, vars["roomID"], syncDB, rsAPI, true, &membership, nil, at)
+		}),
+	).Methods(http.MethodGet, http.MethodOptions)
 }

syncapi/storage/interface.go

@@ -178,6 +178,11 @@ type Database interface {
 	ReIndex(ctx context.Context, limit, afterID int64) (map[int64]gomatrixserverlib.HeaderedEvent, error)
 	UpdateRelations(ctx context.Context, event *gomatrixserverlib.HeaderedEvent) error
 	RedactRelations(ctx context.Context, roomID, redactedEventID string) error
+	SelectMemberships(
+		ctx context.Context,
+		roomID string, pos types.TopologyToken,
+		membership, notMembership *string,
+	) (eventIDs []string, err error)
 }
 
 type Presence interface {

syncapi/storage/postgres/memberships_table.go

@@ -20,11 +20,12 @@ import (
 	"fmt"
 
 	"github.com/lib/pq"
+	"github.com/matrix-org/gomatrixserverlib"
+
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/syncapi/storage/tables"
 	"github.com/matrix-org/dendrite/syncapi/types"
-	"github.com/matrix-org/gomatrixserverlib"
 )
 
 // The memberships table is designed to track the last time that
@@ -69,11 +70,20 @@ const selectHeroesSQL = "" +
 const selectMembershipBeforeSQL = "" +
 	"SELECT membership, topological_pos FROM syncapi_memberships WHERE room_id = $1 and user_id = $2 AND topological_pos <= $3 ORDER BY topological_pos DESC LIMIT 1"
 
+const selectMembersSQL = `
+SELECT event_id FROM (
+	SELECT DISTINCT ON (room_id, user_id) room_id, user_id, event_id, membership FROM syncapi_memberships WHERE room_id = $1 AND topological_pos <= $2 ORDER BY room_id, user_id, stream_pos DESC  
+) t 
+WHERE ($3::text IS NULL OR t.membership = $3)
+	AND ($4::text IS NULL OR t.membership <> $4)
+`
+
 type membershipsStatements struct {
 	upsertMembershipStmt        *sql.Stmt
 	selectMembershipCountStmt   *sql.Stmt
 	selectHeroesStmt            *sql.Stmt
 	selectMembershipForUserStmt *sql.Stmt
+	selectMembersStmt           *sql.Stmt
 }
 
 func NewPostgresMembershipsTable(db *sql.DB) (tables.Memberships, error) {
@@ -87,6 +97,7 @@ func NewPostgresMembershipsTable(db *sql.DB) (tables.Memberships, error) {
 		{&s.selectMembershipCountStmt, selectMembershipCountSQL},
 		{&s.selectHeroesStmt, selectHeroesSQL},
 		{&s.selectMembershipForUserStmt, selectMembershipBeforeSQL},
+		{&s.selectMembersStmt, selectMembersSQL},
 	}.Prepare(db)
 }
 
@@ -154,3 +165,25 @@ func (s *membershipsStatements) SelectMembershipForUser(
 	}
 	return membership, topologyPos, nil
 }
+
+func (s *membershipsStatements) SelectMemberships(
+	ctx context.Context, txn *sql.Tx,
+	roomID string, pos types.TopologyToken,
+	membership, notMembership *string,
+) (eventIDs []string, err error) {
+	stmt := sqlutil.TxStmt(txn, s.selectMembersStmt)
+	rows, err := stmt.QueryContext(ctx, roomID, pos.Depth, membership, notMembership)
+	if err != nil {
+		return
+	}
+	var (
+		eventID string
+	)
+	for rows.Next() {
+		if err = rows.Scan(&eventID); err != nil {
+			return
+		}
+		eventIDs = append(eventIDs, eventID)
+	}
+	return eventIDs, rows.Err()
+}

syncapi/storage/shared/storage_consumer.go

@@ -617,3 +617,11 @@ func (d *Database) RedactRelations(ctx context.Context, roomID, redactedEventID
 		return d.Relations.DeleteRelation(ctx, txn, roomID, redactedEventID)
 	})
 }
+
+func (d *Database) SelectMemberships(
+	ctx context.Context,
+	roomID string, pos types.TopologyToken,
+	membership, notMembership *string,
+) (eventIDs []string, err error) {
+	return d.Memberships.SelectMemberships(ctx, nil, roomID, pos, membership, notMembership)
+}

syncapi/storage/sqlite3/memberships_table.go

@@ -20,11 +20,12 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/matrix-org/gomatrixserverlib"
+
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/syncapi/storage/tables"
 	"github.com/matrix-org/dendrite/syncapi/types"
-	"github.com/matrix-org/gomatrixserverlib"
 )
 
 // The memberships table is designed to track the last time that
@@ -69,12 +70,20 @@ const selectHeroesSQL = "" +
 const selectMembershipBeforeSQL = "" +
 	"SELECT membership, topological_pos FROM syncapi_memberships WHERE room_id = $1 and user_id = $2 AND topological_pos <= $3 ORDER BY topological_pos DESC LIMIT 1"
 
+const selectMembersSQL = `
+SELECT event_id FROM
+ ( SELECT event_id, membership FROM syncapi_memberships WHERE room_id = $1 AND topological_pos <= $2 GROUP BY user_id HAVING(max(stream_pos))) t
+    WHERE ($3 IS NULL OR t.membership = $3)
+		 	AND ($4 IS NULL OR t.membership <> $4)
+`
+
 type membershipsStatements struct {
 	db                        *sql.DB
 	upsertMembershipStmt      *sql.Stmt
 	selectMembershipCountStmt *sql.Stmt
 	//selectHeroesStmt          *sql.Stmt - prepared at runtime due to variadic
 	selectMembershipForUserStmt *sql.Stmt
+	selectMembersStmt           *sql.Stmt
 }
 
 func NewSqliteMembershipsTable(db *sql.DB) (tables.Memberships, error) {
@@ -89,6 +98,7 @@ func NewSqliteMembershipsTable(db *sql.DB) (tables.Memberships, error) {
 		{&s.upsertMembershipStmt, upsertMembershipSQL},
 		{&s.selectMembershipCountStmt, selectMembershipCountSQL},
 		{&s.selectMembershipForUserStmt, selectMembershipBeforeSQL},
+		{&s.selectMembersStmt, selectMembersSQL},
 		// {&s.selectHeroesStmt, selectHeroesSQL}, - prepared at runtime due to variadic
 	}.Prepare(db)
 }
@@ -170,3 +180,23 @@ func (s *membershipsStatements) SelectMembershipForUser(
 	}
 	return membership, topologyPos, nil
 }
+
+func (s *membershipsStatements) SelectMemberships(
+	ctx context.Context, txn *sql.Tx,
+	roomID string, pos types.TopologyToken,
+	membership, notMembership *string,
+) (eventIDs []string, err error) {
+	stmt := sqlutil.TxStmt(txn, s.selectMembersStmt)
+	rows, err := stmt.QueryContext(ctx, roomID, pos.Depth, membership, notMembership)
+	if err != nil {
+		return
+	}
+	var eventID string
+	for rows.Next() {
+		if err = rows.Scan(&eventID); err != nil {
+			return
+		}
+		eventIDs = append(eventIDs, eventID)
+	}
+	return eventIDs, rows.Err()
+}

syncapi/storage/tables/interface.go

@@ -187,6 +187,11 @@ type Memberships interface {
 	SelectMembershipCount(ctx context.Context, txn *sql.Tx, roomID, membership string, pos types.StreamPosition) (count int, err error)
 	SelectHeroes(ctx context.Context, txn *sql.Tx, roomID, userID string, memberships []string) (heroes []string, err error)
 	SelectMembershipForUser(ctx context.Context, txn *sql.Tx, roomID, userID string, pos int64) (membership string, topologicalPos int, err error)
+	SelectMemberships(
+		ctx context.Context, txn *sql.Tx,
+		roomID string, pos types.TopologyToken,
+		membership, notMembership *string,
+	) (eventIDs []string, err error)
 }
 
 type NotificationData interface {

syncapi/streams/stream_pdu.go

@@ -227,14 +227,10 @@ func (p *PDUStreamProvider) addRoomDeltaToResponse(
 	stateFilter *gomatrixserverlib.StateFilter,
 	req *types.SyncRequest,
 ) (types.StreamPosition, error) {
-	if delta.MembershipPos > 0 && delta.Membership == gomatrixserverlib.Leave {
-		// make sure we don't leak recent events after the leave event.
-		// TODO: History visibility makes this somewhat complex to handle correctly. For example:
-		// TODO: This doesn't work for join -> leave in a single /sync request (see events prior to join).
-		// TODO: This will fail on join -> leave -> sensitive msg -> join -> leave
-		//       in a single /sync request
-		// This is all "okay" assuming history_visibility == "shared" which it is by default.
-		r.To = delta.MembershipPos
+
+	originalLimit := eventFilter.Limit
+	if r.Backwards {
+		eventFilter.Limit = int(r.From - r.To)
 	}
 	recentStreamEvents, limited, err := snapshot.RecentEvents(
 		ctx, delta.RoomID, r,
@@ -303,6 +299,12 @@ func (p *PDUStreamProvider) addRoomDeltaToResponse(
 		logrus.WithError(err).Error("unable to apply history visibility filter")
 	}
 
+	if r.Backwards && len(events) > originalLimit {
+		// We're going backwards and the events are ordered chronologically, so take the last `limit` events
+		events = events[len(events)-originalLimit:]
+		limited = true
+	}
+
 	if len(delta.StateEvents) > 0 {
 		updateLatestPosition(delta.StateEvents[len(delta.StateEvents)-1].EventID())
 	}
@@ -473,7 +475,13 @@ func (p *PDUStreamProvider) getJoinResponseForCompleteSync(
 	var prevBatch *types.TopologyToken
 	if len(recentStreamEvents) > 0 {
 		var backwardTopologyPos, backwardStreamPos types.StreamPosition
-		backwardTopologyPos, backwardStreamPos, err = snapshot.PositionInTopology(ctx, recentStreamEvents[0].EventID())
+		event := recentStreamEvents[0]
+		// If this is the beginning of the room, we can't go back further. We're going to return
+		// the TopologyToken from the last event instead. (Synapse returns the /sync next_Batch)
+		if event.Type() == gomatrixserverlib.MRoomCreate && event.StateKeyEquals("") {
+			event = recentStreamEvents[len(recentStreamEvents)-1]
+		}
+		backwardTopologyPos, backwardStreamPos, err = snapshot.PositionInTopology(ctx, event.EventID())
 		if err != nil {
 			return
 		}

syncapi/types/types.go

@@ -234,6 +234,9 @@ func (t *TopologyToken) StreamToken() StreamingToken {
 }
 
 func (t TopologyToken) String() string {
+	if t.Depth <= 0 && t.PDUPosition <= 0 {
+		return ""
+	}
 	return fmt.Sprintf("t%d_%d", t.Depth, t.PDUPosition)
 }
 

sytest-whitelist

@@ -752,4 +752,9 @@ When user joins a room the state is included in the next sync
 When user joins a room the state is included in a gapped sync
 Messages that notify from another user increment notification_count
 Messages that highlight from another user increment unread highlight count
-Notifications can be viewed with GET /notifications
+Notifications can be viewed with GET /notifications
+Can get rooms/{roomId}/messages for a departed room (SPEC-216)
+Local device key changes appear in /keys/changes
+Can get rooms/{roomId}/members at a given point
+Can filter rooms/{roomId}/members
+Current state appears in timeline in private history with many messages after