Merge branch 'master' into ci_multiarch_build_and_push

clientapi/routing/memberships.go

@@ -44,6 +44,13 @@ type joinedMember struct {
 	AvatarURL   string `json:"avatar_url"`
 }
 
+// The database stores 'displayname' without an underscore.
+// Deserialize into this and then change to the actual API response
+type databaseJoinedMember struct {
+	DisplayName string `json:"displayname"`
+	AvatarURL   string `json:"avatar_url"`
+}
+
 // GetMemberships implements GET /rooms/{roomId}/members
 func GetMemberships(
 	req *http.Request, device *userapi.Device, roomID string, joinedOnly bool,
@@ -72,12 +79,12 @@ func GetMemberships(
 		var res getJoinedMembersResponse
 		res.Joined = make(map[string]joinedMember)
 		for _, ev := range queryRes.JoinEvents {
-			var content joinedMember
+			var content databaseJoinedMember
 			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] = content
+			res.Joined[ev.Sender] = joinedMember(content)
 		}
 		return util.JSONResponse{
 			Code: http.StatusOK,

clientapi/routing/peekroom.go

@@ -77,3 +77,28 @@ func PeekRoomByIDOrAlias(
 		}{peekRes.RoomID},
 	}
 }
+
+func UnpeekRoomByID(
+	req *http.Request,
+	device *api.Device,
+	rsAPI roomserverAPI.RoomserverInternalAPI,
+	accountDB accounts.Database,
+	roomID string,
+) util.JSONResponse {
+	unpeekReq := roomserverAPI.PerformUnpeekRequest{
+		RoomID:   roomID,
+		UserID:   device.UserID,
+		DeviceID: device.ID,
+	}
+	unpeekRes := roomserverAPI.PerformUnpeekResponse{}
+
+	rsAPI.PerformUnpeek(req.Context(), &unpeekReq, &unpeekRes)
+	if unpeekRes.Error != nil {
+		return unpeekRes.Error.JSONResponse()
+	}
+
+	return util.JSONResponse{
+		Code: http.StatusOK,
+		JSON: struct{}{},
+	}
+}

clientapi/routing/routing.go

@@ -106,6 +106,9 @@ func Setup(
 	).Methods(http.MethodPost, http.MethodOptions)
 	r0mux.Handle("/peek/{roomIDOrAlias}",
 		httputil.MakeAuthAPI(gomatrixserverlib.Peek, userAPI, func(req *http.Request, device *userapi.Device) util.JSONResponse {
+			if r := rateLimits.rateLimit(req); r != nil {
+				return *r
+			}
 			vars, err := httputil.URLDecodeMapValues(mux.Vars(req))
 			if err != nil {
 				return util.ErrorResponse(err)
@@ -148,6 +151,17 @@ func Setup(
 			)
 		}),
 	).Methods(http.MethodPost, http.MethodOptions)
+	r0mux.Handle("/rooms/{roomID}/unpeek",
+		httputil.MakeAuthAPI("unpeek", 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 UnpeekRoomByID(
+				req, device, rsAPI, accountDB, vars["roomID"],
+			)
+		}),
+	).Methods(http.MethodPost, http.MethodOptions)
 	r0mux.Handle("/rooms/{roomID}/ban",
 		httputil.MakeAuthAPI("membership", userAPI, func(req *http.Request, device *userapi.Device) util.JSONResponse {
 			vars, err := httputil.URLDecodeMapValues(mux.Vars(req))

cmd/create-account/main.go

@@ -20,24 +20,27 @@ import (
 	"fmt"
 	"os"
 
+	"github.com/matrix-org/dendrite/setup"
 	"github.com/matrix-org/dendrite/setup/config"
 	"github.com/matrix-org/dendrite/userapi/storage/accounts"
-	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/sirupsen/logrus"
 )
 
 const usage = `Usage: %s
 
-Generate a new Matrix account for testing purposes.
+Creates a new user account on the homeserver.
+
+Example:
+
+  ./create-account --config dendrite.yaml --username alice --password foobarbaz
 
 Arguments:
 
 `
 
 var (
-	database      = flag.String("database", "", "The location of the account database.")
-	username      = flag.String("username", "", "The user ID localpart to register e.g 'alice' in '@alice:localhost'.")
-	password      = flag.String("password", "", "Optional. The password to register with. If not specified, this account will be password-less.")
-	serverNameStr = flag.String("servername", "localhost", "The Matrix server domain which will form the domain part of the user ID.")
+	username = flag.String("username", "", "The username of the account to register (specify the localpart only, e.g. 'alice' for '@alice:domain.com')")
+	password = flag.String("password", "", "The password to associate with the account (optional, account will be password-less if not specified)")
 )
 
 func main() {
@@ -45,36 +48,24 @@ func main() {
 		fmt.Fprintf(os.Stderr, usage, os.Args[0])
 		flag.PrintDefaults()
 	}
-
-	flag.Parse()
+	cfg := setup.ParseFlags(true)
 
 	if *username == "" {
 		flag.Usage()
-		fmt.Println("Missing --username")
 		os.Exit(1)
 	}
 
-	if *database == "" {
-		flag.Usage()
-		fmt.Println("Missing --database")
-		os.Exit(1)
-	}
-
-	serverName := gomatrixserverlib.ServerName(*serverNameStr)
-
 	accountDB, err := accounts.NewDatabase(&config.DatabaseOptions{
-		ConnectionString: config.DataSource(*database),
-	}, serverName)
+		ConnectionString: cfg.UserAPI.AccountDatabase.ConnectionString,
+	}, cfg.Global.ServerName)
 	if err != nil {
-		fmt.Println(err.Error())
-		os.Exit(1)
+		logrus.Fatalln("Failed to connect to the database:", err.Error())
 	}
 
 	_, err = accountDB.CreateAccount(context.Background(), *username, *password, "")
 	if err != nil {
-		fmt.Println(err.Error())
-		os.Exit(1)
+		logrus.Fatalln("Failed to create the account:", err.Error())
 	}
 
-	fmt.Println("Created account")
+	logrus.Infoln("Created account", *username)
 }

docs/FAQ.md

@@ -0,0 +1,60 @@
+# Frequently Asked Questions
+
+### Is Dendrite stable?
+
+Mostly, although there are still bugs and missing features. If you are a confident power user and you are happy to spend some time debugging things when they go wrong, then please try out Dendrite. If you are a community, organisation or business that demands stability and uptime, then Dendrite is not for you yet - please install Synapse instead.
+
+### Is Dendrite feature-complete?
+
+No, although a good portion of the Matrix specification has been implemented. Mostly missing are client features - see the readme at the root of the repository for more information.
+
+### Is there a migration path from Synapse to Dendrite?
+
+No, not at present. There will be in the future when Dendrite reaches version 1.0.
+
+### I've installed Dendrite but federation isn't working
+
+Check the [Federation Tester](https://federationtester.matrix.org). You need at least:
+
+* A valid DNS name
+* A valid TLS certificate for that DNS name
+* Either DNS SRV records or well-known files
+
+### Does Dendrite work with my favourite client?
+
+It should do, although we are aware of some minor issues:
+
+* **Element Android**: registration does not work, but logging in with an existing account does
+* **Hydrogen**: occasionally sync can fail due to gaps in the `since` parameter, but clearing the cache fixes this
+
+### Does Dendrite support push notifications?
+
+No, not yet. This is a planned feature.
+
+### Does Dendrite support application services/bridges?
+
+Possibly - Dendrite does have some application service support but it is not well tested. Please let us know by raising a GitHub issue if you try it and run into problems.
+
+### Is it possible to prevent communication with the outside world?
+
+Yes, you can do this by disabling federation - set `disable_federation` to `true` in the `global` section of the Dendrite configuration file. 
+
+### Should I use PostgreSQL or SQLite for my databases?
+
+Please use PostgreSQL wherever possible, especially if you are planning to run a homeserver that caters to more than a couple of users. 
+
+### Dendrite is using a lot of CPU
+
+Generally speaking, you should expect to see some CPU spikes, particularly if you are joining or participating in large rooms. However, constant/sustained high CPU usage is not expected - if you are experiencing that, please join `#dendrite-dev:matrix.org` and let us know, or file a GitHub issue.
+
+### Dendrite is using a lot of RAM
+
+A lot of users report that Dendrite is using a lot of RAM, sometimes even gigabytes of it. This is usually due to Go's allocator behaviour, which tries to hold onto allocated memory until the operating system wants to reclaim it for something else. This can make the memory usage look significantly inflated in tools like `top`/`htop` when actually most of that memory is not really in use at all.
+
+If you want to prevent this behaviour so that the Go runtime releases memory normally, start Dendrite using the `GODEBUG=madvdontneed=1` environment variable. It is also expected that the allocator behaviour will be changed again in Go 1.16 so that it does not hold onto memory unnecessarily in this way.
+
+If you are running with `GODEBUG=madvdontneed=1` and still see hugely inflated memory usage then that's quite possibly a bug - please join `#dendrite-dev:matrix.org` and let us know, or file a GitHub issue.
+
+### Dendrite is running out of PostgreSQL database connections
+
+You may need to revisit the connection limit of your PostgreSQL server and/or make changes to the `max_connections` lines in your Dendrite configuration. Be aware that each Dendrite component opens its own database connections and has its own connection limit, even in monolith mode!

docs/PROFILING.md

@@ -0,0 +1,89 @@
+# Profiling Dendrite
+
+If you are running into problems with Dendrite using excessive resources (e.g. CPU or RAM) then you can use the profiler to work out what is happening.
+
+Dendrite contains an embedded profiler called `pprof`, which is a part of the standard Go toolchain. 
+
+## Enable the profiler
+
+To enable the profiler, start Dendrite with the `PPROFLISTEN` environment variable. This variable specifies which address and port to listen on, e.g.
+
+```
+PPROFLISTEN=localhost:65432 ./bin/dendrite-monolith-server ...
+```
+
+If pprof has been enabled successfully, a log line at startup will show that pprof is listening:
+
+```
+WARN[2020-12-03T13:32:33.669405000Z] [/Users/neilalexander/Desktop/dendrite/internal/log.go:87] SetupPprof
+	 Starting pprof on localhost:65432
+```
+
+All examples from this point forward assume `PPROFLISTEN=localhost:65432` but you may need to adjust as necessary for your setup.
+
+## Profiling CPU usage
+
+To examine where CPU time is going, you can call the `profile` endpoint:
+
+```
+http://localhost:65432/debug/pprof/profile?seconds=30
+```
+
+The profile will run for the specified number of `seconds` and then will produce a result.
+
+### Examine a profile using the Go toolchain
+
+If you have Go installed and want to explore the profile, you can invoke `go tool pprof` to start the profile directly. The `-http=` parameter will instruct `go tool pprof` to start a web server providing a view of the captured profile:
+
+```
+go tool pprof -http=localhost:23456 http://localhost:65432/debug/pprof/profile?seconds=30
+```
+
+You can then visit `http://localhost:23456` in your web browser to see a visual representation of the profile. Particularly usefully, in the "View" menu, you can select "Flame Graph" to see a proportional interactive graph of CPU usage.
+
+### Download a profile to send to someone else
+
+If you don't have the Go tools installed but just want to capture the profile to send to someone else, you can instead use `curl` to download the profiler results:
+
+```
+curl -O http://localhost:65432/debug/pprof/profile?seconds=30
+```
+
+This will block for the specified number of seconds, capturing information about what Dendrite is doing, and then produces a `profile` file, which you can send onward.
+
+## Profiling memory usage
+
+To examine where memory usage is going, you can call the `heap` endpoint:
+
+```
+http://localhost:65432/debug/pprof/heap
+```
+
+The profile will return almost instantly.
+
+### Examine a profile using the Go toolchain
+
+If you have Go installed and want to explore the profile, you can invoke `go tool pprof` to start the profile directly. The `-http=` parameter will instruct `go tool pprof` to start a web server providing a view of the captured profile:
+
+```
+go tool pprof -http=localhost:23456 http://localhost:65432/debug/pprof/heap
+```
+
+You can then visit `http://localhost:23456` in your web browser to see a visual representation of the profile. The "Sample" menu lets you select between four different memory profiles:
+
+* `inuse_space`: Shows how much actual heap memory is allocated per function (this is generally the most useful profile when diagnosing high memory usage)
+* `inuse_objects`: Shows how many heap objects are allocated per function
+* `alloc_space`: Shows how much memory has been allocated per function (although that memory may have since been deallocated)
+* `alloc_objects`: Shows how many allocations have been made per function (although that memory may have since been deallocated)
+
+Also in the "View" menu, you can select "Flame Graph" to see a proportional interactive graph of the memory usage.
+
+### Download a profile to send to someone else
+
+If you don't have the Go tools installed but just want to capture the profile to send to someone else, you can instead use `curl` to download the profiler results:
+
+```
+curl -O http://localhost:65432/debug/pprof/heap
+```
+
+This will almost instantly produce a `heap` file, which you can send onward.

docs/hiawatha/monolith-sample.conf

@@ -0,0 +1,17 @@
+# Depending on which port is used for federation (.well-known/matrix/server or SRV record),
+# ensure there's a binding for that port in the configuration. Replace "FEDPORT" with port
+# number, (e.g. "8448"), and "IPV4" with your server's ipv4 address (separate binding for
+# each ip address, e.g. if you use both ipv4 and ipv6 addresses).
+
+Binding {
+        Port = FEDPORT
+        Interface = IPV4
+        TLScertFile = /path/to/fullchainandprivkey.pem
+}
+
+VirtualHost {
+        ...
+        ReverseProxy = /_matrix http://localhost:8008 600
+        ...
+
+}

docs/hiawatha/polylith-sample.conf

@@ -1,3 +1,15 @@
+# Depending on which port is used for federation (.well-known/matrix/server or SRV record),
+# ensure there's a binding for that port in the configuration. Replace "FEDPORT" with port
+# number, (e.g. "8448"), and "IPV4" with your server's ipv4 address (separate binding for
+# each ip address, e.g. if you use both ipv4 and ipv6 addresses).
+
+Binding {
+        Port = FEDPORT
+        Interface = IPV4
+        TLScertFile = /path/to/fullchainandprivkey.pem
+}
+
+
 VirtualHost {
         ...
         # route requests to:
@@ -7,10 +19,10 @@ VirtualHost {
         # /_matrix/client/.*/keys/changes
         # /_matrix/client/.*/rooms/{roomId}/messages
         # to sync_api
-        ReverseProxy = /_matrix/client/.*?/(sync|user/.*?/filter/?.*|keys/changes|rooms/.*?/messages) http://localhost:8073
-        ReverseProxy = /_matrix/client http://localhost:8071
-        ReverseProxy = /_matrix/federation http://localhost:8072
-        ReverseProxy = /_matrix/key http://localhost:8072
-        ReverseProxy = /_matrix/media http://localhost:8074
+        ReverseProxy = /_matrix/client/.*?/(sync|user/.*?/filter/?.*|keys/changes|rooms/.*?/messages) 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
+        ReverseProxy = /_matrix/media http://localhost:8074 600
         ...
 }

federationapi/routing/send_test.go

@@ -131,6 +131,13 @@ func (t *testRoomserverAPI) PerformPeek(
 ) {
 }
 
+func (t *testRoomserverAPI) PerformUnpeek(
+	ctx context.Context,
+	req *api.PerformUnpeekRequest,
+	res *api.PerformUnpeekResponse,
+) {
+}
+
 func (t *testRoomserverAPI) PerformPublish(
 	ctx context.Context,
 	req *api.PerformPublishRequest,

roomserver/api/api.go

@@ -42,6 +42,12 @@ type RoomserverInternalAPI interface {
 		res *PerformPeekResponse,
 	)
 
+	PerformUnpeek(
+		ctx context.Context,
+		req *PerformUnpeekRequest,
+		res *PerformUnpeekResponse,
+	)
+
 	PerformPublish(
 		ctx context.Context,
 		req *PerformPublishRequest,

roomserver/api/api_trace.go

@@ -46,6 +46,15 @@ func (t *RoomserverInternalAPITrace) PerformPeek(
 	util.GetLogger(ctx).Infof("PerformPeek req=%+v res=%+v", js(req), js(res))
 }
 
+func (t *RoomserverInternalAPITrace) PerformUnpeek(
+	ctx context.Context,
+	req *PerformUnpeekRequest,
+	res *PerformUnpeekResponse,
+) {
+	t.Impl.PerformUnpeek(ctx, req, res)
+	util.GetLogger(ctx).Infof("PerformUnpeek req=%+v res=%+v", js(req), js(res))
+}
+
 func (t *RoomserverInternalAPITrace) PerformJoin(
 	ctx context.Context,
 	req *PerformJoinRequest,

roomserver/api/output.go

@@ -51,6 +51,8 @@ const (
 
 	// OutputTypeNewPeek indicates that the kafka event is an OutputNewPeek
 	OutputTypeNewPeek OutputType = "new_peek"
+	// OutputTypeRetirePeek indicates that the kafka event is an OutputRetirePeek
+	OutputTypeRetirePeek OutputType = "retire_peek"
 )
 
 // An OutputEvent is an entry in the roomserver output kafka log.
@@ -70,6 +72,8 @@ type OutputEvent struct {
 	RedactedEvent *OutputRedactedEvent `json:"redacted_event,omitempty"`
 	// The content of event with type OutputTypeNewPeek
 	NewPeek *OutputNewPeek `json:"new_peek,omitempty"`
+	// The content of event with type OutputTypeRetirePeek
+	RetirePeek *OutputRetirePeek `json:"retire_peek,omitempty"`
 }
 
 // Type of the OutputNewRoomEvent.
@@ -240,3 +244,10 @@ type OutputNewPeek struct {
 	UserID   string
 	DeviceID string
 }
+
+// An OutputRetirePeek is written whenever a user stops peeking into a room.
+type OutputRetirePeek struct {
+	RoomID   string
+	UserID   string
+	DeviceID string
+}

roomserver/api/perform.go

@@ -123,6 +123,17 @@ type PerformPeekResponse struct {
 	Error *PerformError
 }
 
+type PerformUnpeekRequest struct {
+	RoomID   string `json:"room_id"`
+	UserID   string `json:"user_id"`
+	DeviceID string `json:"device_id"`
+}
+
+type PerformUnpeekResponse struct {
+	// If non-nil, the join request failed. Contains more information why it failed.
+	Error *PerformError
+}
+
 // PerformBackfillRequest is a request to PerformBackfill.
 type PerformBackfillRequest struct {
 	// The room to backfill

roomserver/internal/api.go

@@ -23,6 +23,7 @@ type RoomserverInternalAPI struct {
 	*perform.Inviter
 	*perform.Joiner
 	*perform.Peeker
+	*perform.Unpeeker
 	*perform.Leaver
 	*perform.Publisher
 	*perform.Backfiller
@@ -94,6 +95,13 @@ func (r *RoomserverInternalAPI) SetFederationSenderAPI(fsAPI fsAPI.FederationSen
 		FSAPI:      r.fsAPI,
 		Inputer:    r.Inputer,
 	}
+	r.Unpeeker = &perform.Unpeeker{
+		ServerName: r.Cfg.Matrix.ServerName,
+		Cfg:        r.Cfg,
+		DB:         r.DB,
+		FSAPI:      r.fsAPI,
+		Inputer:    r.Inputer,
+	}
 	r.Leaver = &perform.Leaver{
 		Cfg:     r.Cfg,
 		DB:      r.DB,

roomserver/internal/perform/perform_peek.go

@@ -163,8 +163,7 @@ func (r *Peeker) performPeekRoomByID(
 	// XXX: we should probably factor out history_visibility checks into a common utility method somewhere
 	// which handles the default value etc.
 	var worldReadable = false
-	ev, _ := r.DB.GetStateEvent(ctx, roomID, "m.room.history_visibility", "")
-	if ev != nil {
+	if ev, _ := r.DB.GetStateEvent(ctx, roomID, "m.room.history_visibility", ""); ev != nil {
 		content := map[string]string{}
 		if err = json.Unmarshal(ev.Content(), &content); err != nil {
 			util.GetLogger(ctx).WithError(err).Error("json.Unmarshal for history visibility failed")
@@ -182,6 +181,13 @@ func (r *Peeker) performPeekRoomByID(
 		}
 	}
 
+	if ev, _ := r.DB.GetStateEvent(ctx, roomID, "m.room.encryption", ""); ev != nil {
+		return "", &api.PerformError{
+			Code: api.PerformErrorNotAllowed,
+			Msg:  "Cannot peek into an encrypted room",
+		}
+	}
+
 	// TODO: handle federated peeks
 
 	err = r.Inputer.WriteOutputEvents(roomID, []api.OutputEvent{

roomserver/internal/perform/perform_unpeek.go

@@ -0,0 +1,118 @@
+// Copyright 2020 New Vector Ltd
+//
+// 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 perform
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	fsAPI "github.com/matrix-org/dendrite/federationsender/api"
+	"github.com/matrix-org/dendrite/roomserver/api"
+	"github.com/matrix-org/dendrite/roomserver/internal/input"
+	"github.com/matrix-org/dendrite/roomserver/storage"
+	"github.com/matrix-org/dendrite/setup/config"
+	"github.com/matrix-org/gomatrixserverlib"
+)
+
+type Unpeeker struct {
+	ServerName gomatrixserverlib.ServerName
+	Cfg        *config.RoomServer
+	FSAPI      fsAPI.FederationSenderInternalAPI
+	DB         storage.Database
+
+	Inputer *input.Inputer
+}
+
+// PerformPeek handles peeking into matrix rooms, including over federation by talking to the federationsender.
+func (r *Unpeeker) PerformUnpeek(
+	ctx context.Context,
+	req *api.PerformUnpeekRequest,
+	res *api.PerformUnpeekResponse,
+) {
+	if err := r.performUnpeek(ctx, req); err != nil {
+		perr, ok := err.(*api.PerformError)
+		if ok {
+			res.Error = perr
+		} else {
+			res.Error = &api.PerformError{
+				Msg: err.Error(),
+			}
+		}
+	}
+}
+
+func (r *Unpeeker) performUnpeek(
+	ctx context.Context,
+	req *api.PerformUnpeekRequest,
+) error {
+	// FIXME: there's way too much duplication with performJoin
+	_, domain, err := gomatrixserverlib.SplitID('@', req.UserID)
+	if err != nil {
+		return &api.PerformError{
+			Code: api.PerformErrorBadRequest,
+			Msg:  fmt.Sprintf("Supplied user ID %q in incorrect format", req.UserID),
+		}
+	}
+	if domain != r.Cfg.Matrix.ServerName {
+		return &api.PerformError{
+			Code: api.PerformErrorBadRequest,
+			Msg:  fmt.Sprintf("User %q does not belong to this homeserver", req.UserID),
+		}
+	}
+	if strings.HasPrefix(req.RoomID, "!") {
+		return r.performUnpeekRoomByID(ctx, req)
+	}
+	return &api.PerformError{
+		Code: api.PerformErrorBadRequest,
+		Msg:  fmt.Sprintf("Room ID %q is invalid", req.RoomID),
+	}
+}
+
+func (r *Unpeeker) performUnpeekRoomByID(
+	_ context.Context,
+	req *api.PerformUnpeekRequest,
+) (err error) {
+	// Get the domain part of the room ID.
+	_, _, err = gomatrixserverlib.SplitID('!', req.RoomID)
+	if err != nil {
+		return &api.PerformError{
+			Code: api.PerformErrorBadRequest,
+			Msg:  fmt.Sprintf("Room ID %q is invalid: %s", req.RoomID, err),
+		}
+	}
+
+	// TODO: handle federated peeks
+
+	err = r.Inputer.WriteOutputEvents(req.RoomID, []api.OutputEvent{
+		{
+			Type: api.OutputTypeRetirePeek,
+			RetirePeek: &api.OutputRetirePeek{
+				RoomID:   req.RoomID,
+				UserID:   req.UserID,
+				DeviceID: req.DeviceID,
+			},
+		},
+	})
+	if err != nil {
+		return
+	}
+
+	// By this point, if req.RoomIDOrAlias contained an alias, then
+	// it will have been overwritten with a room ID by performPeekRoomByAlias.
+	// We should now include this in the response so that the CS API can
+	// return the right room ID.
+	return nil
+}

roomserver/inthttp/client.go

@@ -27,6 +27,7 @@ const (
 	// Perform operations
 	RoomserverPerformInvitePath   = "/roomserver/performInvite"
 	RoomserverPerformPeekPath     = "/roomserver/performPeek"
+	RoomserverPerformUnpeekPath   = "/roomserver/performUnpeek"
 	RoomserverPerformJoinPath     = "/roomserver/performJoin"
 	RoomserverPerformLeavePath    = "/roomserver/performLeave"
 	RoomserverPerformBackfillPath = "/roomserver/performBackfill"
@@ -209,6 +210,23 @@ func (h *httpRoomserverInternalAPI) PerformPeek(
 	}
 }
 
+func (h *httpRoomserverInternalAPI) PerformUnpeek(
+	ctx context.Context,
+	request *api.PerformUnpeekRequest,
+	response *api.PerformUnpeekResponse,
+) {
+	span, ctx := opentracing.StartSpanFromContext(ctx, "PerformUnpeek")
+	defer span.Finish()
+
+	apiURL := h.roomserverURL + RoomserverPerformUnpeekPath
+	err := httputil.PostJSON(ctx, span, h.httpClient, apiURL, request, response)
+	if err != nil {
+		response.Error = &api.PerformError{
+			Msg: fmt.Sprintf("failed to communicate with roomserver: %s", err),
+		}
+	}
+}
+
 func (h *httpRoomserverInternalAPI) PerformLeave(
 	ctx context.Context,
 	request *api.PerformLeaveRequest,

roomserver/inthttp/server.go

@@ -72,6 +72,17 @@ func AddRoutes(r api.RoomserverInternalAPI, internalAPIMux *mux.Router) {
 			return util.JSONResponse{Code: http.StatusOK, JSON: &response}
 		}),
 	)
+	internalAPIMux.Handle(RoomserverPerformPeekPath,
+		httputil.MakeInternalAPI("performUnpeek", func(req *http.Request) util.JSONResponse {
+			var request api.PerformUnpeekRequest
+			var response api.PerformUnpeekResponse
+			if err := json.NewDecoder(req.Body).Decode(&request); err != nil {
+				return util.MessageResponse(http.StatusBadRequest, err.Error())
+			}
+			r.PerformUnpeek(req.Context(), &request, &response)
+			return util.JSONResponse{Code: http.StatusOK, JSON: &response}
+		}),
+	)
 	internalAPIMux.Handle(RoomserverPerformPublishPath,
 		httputil.MakeInternalAPI("performPublish", func(req *http.Request) util.JSONResponse {
 			var request api.PerformPublishRequest

syncapi/consumers/roomserver.go

@@ -105,6 +105,8 @@ func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error {
 		return s.onRetireInviteEvent(context.TODO(), *output.RetireInviteEvent)
 	case api.OutputTypeNewPeek:
 		return s.onNewPeek(context.TODO(), *output.NewPeek)
+	case api.OutputTypeRetirePeek:
+		return s.onRetirePeek(context.TODO(), *output.RetirePeek)
 	case api.OutputTypeRedactedEvent:
 		return s.onRedactEvent(context.TODO(), *output.RedactedEvent)
 	default:
@@ -309,6 +311,26 @@ func (s *OutputRoomEventConsumer) onNewPeek(
 	return nil
 }
 
+func (s *OutputRoomEventConsumer) onRetirePeek(
+	ctx context.Context, msg api.OutputRetirePeek,
+) error {
+	sp, err := s.db.DeletePeek(ctx, msg.RoomID, msg.UserID, msg.DeviceID)
+	if err != nil {
+		// panic rather than continue with an inconsistent database
+		log.WithFields(log.Fields{
+			log.ErrorKey: err,
+		}).Panicf("roomserver output log: write peek failure")
+		return nil
+	}
+	// tell the notifier about the new peek so it knows to wake up new devices
+	s.notifier.OnRetirePeek(msg.RoomID, msg.UserID, msg.DeviceID)
+
+	// we need to wake up the users who might need to now be peeking into this room,
+	// so we send in a dummy event to trigger a wakeup
+	s.notifier.OnNewEvent(nil, msg.RoomID, nil, types.NewStreamToken(sp, 0, nil))
+	return nil
+}
+
 func (s *OutputRoomEventConsumer) updateStateEvent(event *gomatrixserverlib.HeaderedEvent) (*gomatrixserverlib.HeaderedEvent, error) {
 	if event.StateKey() == nil {
 		return event, nil

syncapi/storage/interface.go

@@ -91,6 +91,9 @@ type Database interface {
 	// AddPeek adds a new peek to our DB for a given room by a given user's device.
 	// Returns an error if there was a problem communicating with the database.
 	AddPeek(ctx context.Context, RoomID, UserID, DeviceID string) (types.StreamPosition, error)
+	// DeletePeek removes an existing peek from the database for a given room by a user's device.
+	// Returns an error if there was a problem communicating with the database.
+	DeletePeek(ctx context.Context, roomID, userID, deviceID string) (sp types.StreamPosition, err error)
 	// DeletePeek deletes all peeks for a given room by a given user
 	// Returns an error if there was a problem communicating with the database.
 	DeletePeeks(ctx context.Context, RoomID, UserID string) (types.StreamPosition, error)

syncapi/storage/shared/syncserver.go

@@ -178,6 +178,23 @@ func (d *Database) AddPeek(
 	return
 }
 
+// DeletePeeks tracks the fact that a user has stopped peeking from the specified
+// device. If the peeks was successfully deleted this returns the stream ID it was
+// stored at. Returns an error if there was a problem communicating with the database.
+func (d *Database) DeletePeek(
+	ctx context.Context, roomID, userID, deviceID string,
+) (sp types.StreamPosition, err error) {
+	err = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
+		sp, err = d.Peeks.DeletePeek(ctx, txn, roomID, userID, deviceID)
+		return err
+	})
+	if err == sql.ErrNoRows {
+		sp = 0
+		err = nil
+	}
+	return
+}
+
 // DeletePeeks tracks the fact that a user has stopped peeking from all devices
 // If the peeks was successfully deleted this returns the stream ID it was stored at.
 // Returns an error if there was a problem communicating with the database.

syncapi/sync/notifier.go

@@ -137,6 +137,18 @@ func (n *Notifier) OnNewPeek(
 	// by calling OnNewEvent.
 }
 
+func (n *Notifier) OnRetirePeek(
+	roomID, userID, deviceID string,
+) {
+	n.streamLock.Lock()
+	defer n.streamLock.Unlock()
+
+	n.removePeekingDevice(roomID, userID, deviceID)
+
+	// we don't wake up devices here given the roomserver consumer will do this shortly afterwards
+	// by calling OnRetireEvent.
+}
+
 func (n *Notifier) OnNewSendToDevice(
 	userID string, deviceIDs []string,
 	posUpdate types.StreamingToken,

sytest-whitelist

@@ -503,3 +503,4 @@ Forgetting room does not show up in v2 /sync
 Can forget room you've been kicked from
 Can re-join room if re-invited
 /whois
+/joined_members return joined members