Merge v0.8.9 (#13)

Squashed commit of the following:

commit b5c55faf98
Author: Neil Alexander <neilalexander@users.noreply.github.com>
Date:   Fri Jul 1 12:00:32 2022 +0100

    Version 0.8.9 (#2549)

    * Version 0.8.9

    * Update changelog

commit b50a24c666
Author: Neil Alexander <neilalexander@users.noreply.github.com>
Date:   Fri Jul 1 10:54:07 2022 +0100

    Roomserver producers package (#2546)

    * Give the roomserver a producers package

    * Change init point

    * Populate ACLs API

    * Fix build issues

    * `RoomEventProducer` naming

commit 89cd0e8fc1
Author: Till <2353100+S7evinK@users.noreply.github.com>
Date:   Fri Jul 1 11:49:26 2022 +0200

    Try to fix backfilling (#2548)

    * Try to fix backfilling

    * Return start/end to not confuse clients

    * Update GMSL

    * Update GMSL

commit 086f182e24
Author: Neil Alexander <neilalexander@users.noreply.github.com>
Date:   Fri Jul 1 09:50:06 2022 +0100

    Disable WebAssembly builds for now

commit 54bed4c593
Author: Neil Alexander <neilalexander@users.noreply.github.com>
Date:   Fri Jul 1 09:37:54 2022 +0100

    Blacklist `Guest users can join guest_access rooms` test until it can be investigated

commit 561c159ad7
Author: Till <2353100+S7evinK@users.noreply.github.com>
Date:   Thu Jun 30 12:34:37 2022 +0200

    Silence presence logs (#2547)

commit 519bc1124b
Author: Neil Alexander <neilalexander@users.noreply.github.com>
Date:   Wed Jun 29 15:29:39 2022 +0100

    Add `evacuateUser` endpoint, use it when deactivating accounts (#2545)

    * Add `evacuateUser` endpoint, use it when deactivating accounts

    * Populate the API

    * Clean up user devices when deactivating

    * Include invites, delete pushers

commit 2dea466685
Author: Neil Alexander <neilalexander@users.noreply.github.com>
Date:   Wed Jun 29 12:32:24 2022 +0100

    Return an error if trying to invite a malformed user ID (#2543)

commit 2086992caf
Author: Till <2353100+S7evinK@users.noreply.github.com>
Date:   Wed Jun 29 10:49:12 2022 +0200

    Don't return `end` if there are not more messages (#2542)

    * Be more spec compliant

    * Move lazyLoadMembers to own method

commit 920a20821b
Author: Jean Lucas <jean@4ray.co>
Date:   Mon Jun 27 04:15:19 2022 -0400

    Fix nats.go commit (#2540)

    Signed-off-by: Jean Lucas <jean@4ray.co>

commit 7120eb6bc9
Author: Neil Alexander <neilalexander@users.noreply.github.com>
Date:   Wed Jun 15 14:27:07 2022 +0100

    Add `InputDeviceListUpdate` to the keyserver, remove old input API (#2536)

    * Add `InputDeviceListUpdate` to the keyserver, remove old input API

    * Fix copyright

    * Log more information when a device list update fails

commit 1b90cc9536
Author: Till <2353100+S7evinK@users.noreply.github.com>
Date:   Wed Jun 15 12:50:02 2022 +0200

    Fix rare panic when returning user devices over federation (#2534)

commit 4c2a10f1a6
Author: Neil Alexander <neilalexander@users.noreply.github.com>
Date:   Mon Jun 13 15:11:10 2022 +0100

    Handle state before, send history visibility in output (#2532)

    * Check state before event

    * Tweaks

    * Refactor a bit, include in output events

    * Don't waste time if soft failed either

    * Tweak control flow, comments, use GMSL history visibility type

commit c500958583
Author: Emanuele Aliberti <dev@mtka.eu>
Date:   Mon Jun 13 13:08:46 2022 +0200

    generic CaddyFile in front of Dendrite (monolith) (#2531)

    for Caddy 2.5.x

    Co-authored-by: emanuele.aliberti <emanuele.aliberti@mtka.eu>

commit e1136f4d3e
Author: Till Faelligen <davidf@element.io>
Date:   Mon Jun 13 11:46:59 2022 +0200

    Make the linter happy again

commit 0a7f7dc716
Author: Neil Alexander <neilalexander@users.noreply.github.com>
Date:   Mon Jun 13 10:16:30 2022 +0100

    Add `--difference` to `resolve-state` tool

commit 89d2adadbd
Author: Neil Alexander <neilalexander@users.noreply.github.com>
Date:   Fri Jun 10 10:58:04 2022 +0100

    Attempt to raise the file descriptor limit at startup (#2527)

commit 1030072285
Author: Neil Alexander <neilalexander@users.noreply.github.com>
Date:   Fri Jun 10 10:18:32 2022 +0100

    Rename the page to "Optimise your installation"

commit 16ed1633b6
Author: Neil Alexander <neilalexander@users.noreply.github.com>
Date:   Fri Jun 10 10:15:14 2022 +0100

    Highlighting in docs

commit e2a64773ce
Author: Neil Alexander <neilalexander@users.noreply.github.com>
Date:   Fri Jun 10 10:14:15 2022 +0100

    Add new next steps page to the documentation

commit 660f7839f5
Author: Till <2353100+S7evinK@users.noreply.github.com>
Date:   Thu Jun 9 18:38:07 2022 +0200

    Correctly redact events over federation (#2526)

    * Ensure we check powerlevel/origin before redacting an event

    * Add passing test

    * Use pl.UserLevel

    * Make check more readable, also check for the sender

CHANGES.md

@@ -1,5 +1,22 @@
 # Changelog
 
+## Dendrite 0.8.9 (2022-07-01)
+
+### Features
+
+* Incoming device list updates over federation are now queued in JetStream for processing so that they will no longer block incoming federation transactions and should never end up dropped, which will hopefully help E2EE reliability
+* The `/context` endpoint now returns `"start"` and `"end"` parameters to allow pagination from a context call
+* The `/messages` endpoint will no longer return `"end"` when there are no more messages remaining
+* Deactivated user accounts will now leave all rooms automatically
+* New admin endpoint `/_dendrite/admin/evacuateUser/{userID}` has been added for forcing a local user to leave all joined rooms
+* Dendrite will now automatically attempt to raise the file descriptor limit at startup if it is too low
+
+### Fixes
+
+* A rare crash when retrieving remote device lists has been fixed
+* Fixes a bug where events were not redacted properly over federation
+* The `/invite` endpoints will now return an error instead of silently proceeding if the user ID is obviously malformed
+
 ## Dendrite 0.8.8 (2022-06-09)
 
 ### Features

build.sh

@@ -21,4 +21,4 @@ mkdir -p bin
 
 CGO_ENABLED=1 go build -trimpath -ldflags "$FLAGS" -v -o "bin/" ./cmd/...
 
-CGO_ENABLED=0 GOOS=js GOARCH=wasm go build -trimpath -ldflags "$FLAGS" -o bin/main.wasm ./cmd/dendritejs-pinecone
+# CGO_ENABLED=0 GOOS=js GOARCH=wasm go build -trimpath -ldflags "$FLAGS" -o bin/main.wasm ./cmd/dendritejs-pinecone

clientapi/routing/admin.go

@@ -47,3 +47,40 @@ func AdminEvacuateRoom(req *http.Request, device *userapi.Device, rsAPI roomserv
 		},
 	}
 }
+
+func AdminEvacuateUser(req *http.Request, device *userapi.Device, rsAPI roomserverAPI.ClientRoomserverAPI) util.JSONResponse {
+	if device.AccountType != userapi.AccountTypeAdmin {
+		return util.JSONResponse{
+			Code: http.StatusForbidden,
+			JSON: jsonerror.Forbidden("This API can only be used by admin users."),
+		}
+	}
+	vars, err := httputil.URLDecodeMapValues(mux.Vars(req))
+	if err != nil {
+		return util.ErrorResponse(err)
+	}
+	userID, ok := vars["userID"]
+	if !ok {
+		return util.JSONResponse{
+			Code: http.StatusBadRequest,
+			JSON: jsonerror.MissingArgument("Expecting user ID."),
+		}
+	}
+	res := &roomserverAPI.PerformAdminEvacuateUserResponse{}
+	rsAPI.PerformAdminEvacuateUser(
+		req.Context(),
+		&roomserverAPI.PerformAdminEvacuateUserRequest{
+			UserID: userID,
+		},
+		res,
+	)
+	if err := res.Error; err != nil {
+		return err.JSONResponse()
+	}
+	return util.JSONResponse{
+		Code: 200,
+		JSON: map[string]interface{}{
+			"affected": res.Affected,
+		},
+	}
+}

clientapi/routing/aliases.go

@@ -44,7 +44,7 @@ func GetAliases(
 		return util.ErrorResponse(fmt.Errorf("rsAPI.QueryCurrentState: %w", err))
 	}
 
-	visibility := "invite"
+	visibility := gomatrixserverlib.HistoryVisibilityInvited
 	if historyVisEvent, ok := stateRes.StateEvents[stateTuple]; ok {
 		var err error
 		visibility, err = historyVisEvent.HistoryVisibility()

clientapi/routing/routing.go

@@ -131,6 +131,12 @@ func Setup(
 		}),
 	).Methods(http.MethodGet, http.MethodOptions)
 
+	dendriteAdminRouter.Handle("/admin/evacuateUser/{userID}",
+		httputil.MakeAuthAPI("admin_evacuate_user", userAPI, func(req *http.Request, device *userapi.Device) util.JSONResponse {
+			return AdminEvacuateUser(req, device, rsAPI)
+		}),
+	).Methods(http.MethodGet, http.MethodOptions)
+
 	// server notifications
 	if cfg.Matrix.ServerNotices.Enabled {
 		logrus.Info("Enabling server notices at /_synapse/admin/v1/send_server_notice")

cmd/resolve-state/main.go

@@ -28,7 +28,9 @@ import (
 
 var roomVersion = flag.String("roomversion", "5", "the room version to parse events as")
 var filterType = flag.String("filtertype", "", "the event types to filter on")
+var difference = flag.Bool("difference", false, "whether to calculate the difference between snapshots")
 
+// nolint:gocyclo
 func main() {
 	ctx := context.Background()
 	cfg := setup.ParseFlags(true)
@@ -36,6 +38,7 @@ func main() {
 		Type:  "std",
 		Level: "error",
 	})
+	cfg.ClientAPI.RegistrationDisabled = true
 	base := base.NewBaseDendrite(cfg, "ResolveState", base.DisableMetrics)
 	args := flag.Args()
 
@@ -64,6 +67,59 @@ func main() {
 		RoomVersion: gomatrixserverlib.RoomVersion(*roomVersion),
 	})
 
+	if *difference {
+		if len(snapshotNIDs) != 2 {
+			panic("need exactly two state snapshot NIDs to calculate difference")
+		}
+		var removed, added []types.StateEntry
+		removed, added, err = stateres.DifferenceBetweeenStateSnapshots(ctx, snapshotNIDs[0], snapshotNIDs[1])
+		if err != nil {
+			panic(err)
+		}
+
+		var eventNIDs []types.EventNID
+		for _, entry := range append(removed, added...) {
+			eventNIDs = append(eventNIDs, entry.EventNID)
+		}
+
+		var eventEntries []types.Event
+		eventEntries, err = roomserverDB.Events(ctx, eventNIDs)
+		if err != nil {
+			panic(err)
+		}
+
+		events := make(map[types.EventNID]*gomatrixserverlib.Event, len(eventEntries))
+		for _, entry := range eventEntries {
+			events[entry.EventNID] = entry.Event
+		}
+
+		if len(removed) > 0 {
+			fmt.Println("Removed:")
+			for _, r := range removed {
+				event := events[r.EventNID]
+				fmt.Println()
+				fmt.Printf("* %s %s %q\n", event.EventID(), event.Type(), *event.StateKey())
+				fmt.Printf("  %s\n", string(event.Content()))
+			}
+		}
+
+		if len(removed) > 0 && len(added) > 0 {
+			fmt.Println()
+		}
+
+		if len(added) > 0 {
+			fmt.Println("Added:")
+			for _, a := range added {
+				event := events[a.EventNID]
+				fmt.Println()
+				fmt.Printf("* %s %s %q\n", event.EventID(), event.Type(), *event.StateKey())
+				fmt.Printf("  %s\n", string(event.Content()))
+			}
+		}
+
+		return
+	}
+
 	var stateEntries []types.StateEntry
 	for _, snapshotNID := range snapshotNIDs {
 		var entries []types.StateEntry

docs/administration/4_adminapi.md

@@ -19,6 +19,12 @@ This endpoint will instruct Dendrite to part all local users from the given `roo
 in the URL. It may take some time to complete. A JSON body will be returned containing
 the user IDs of all affected users.
 
+## `/_dendrite/admin/evacuateUser/{userID}`
+
+This endpoint will instruct Dendrite to part the given local `userID` in the URL from
+all rooms which they are currently joined. A JSON body will be returned containing
+the room IDs of all affected rooms.
+
 ## `/_synapse/admin/v1/register`
 
 Shared secret registration — please see the [user creation page](createusers) for

docs/caddy/monolith/CaddyFile

@@ -0,0 +1,68 @@
+{
+    # debug
+    admin off
+    email example@example.com
+    default_sni example.com
+    # Debug endpoint
+    # acme_ca https://acme-staging-v02.api.letsencrypt.org/directory	
+}
+
+#######################################################################
+#   Snippets
+#______________________________________________________________________
+
+(handle_errors_maintenance) {
+	handle_errors {
+		@maintenance expression {http.error.status_code} == 502
+		rewrite @maintenance maintenance.html
+		root * "/path/to/service/pages"
+		file_server
+	}
+}
+
+(matrix-well-known-header) {
+    # Headers
+    header Access-Control-Allow-Origin "*"
+    header Access-Control-Allow-Methods "GET, POST, PUT, DELETE, OPTIONS"
+    header Access-Control-Allow-Headers "Origin, X-Requested-With, Content-Type, Accept, Authorization"
+    header Content-Type "application/json"
+}
+
+#######################################################################
+
+example.com {
+
+	# ...
+
+	handle /.well-known/matrix/server {
+		import matrix-well-known-header
+		respond `{ "m.server": "matrix.example.com:443" }` 200
+	}
+	
+	handle /.well-known/matrix/client {
+		import matrix-well-known-header
+		respond `{ "m.homeserver": { "base_url": "https://matrix.example.com" } }` 200
+	}
+
+	import handle_errors_maintenance
+}
+
+example.com:8448 {
+	# server<->server HTTPS traffic
+    reverse_proxy http://dendrite-host:8008
+}
+
+matrix.example.com {
+
+	handle /_matrix/* {
+		# client<->server HTTPS traffic
+		reverse_proxy  http://dendrite-host:8008
+	}
+
+	handle_path /* {
+		# Client webapp (Element SPA or ...)
+		file_server {
+			root /path/to/www/example.com/matrix-web-client/
+		}	
+	}
+}

docs/installation/10_optimisation.md

@@ -0,0 +1,71 @@
+---
+title: Optimise your installation
+parent: Installation
+has_toc: true
+nav_order: 10
+permalink: /installation/start/optimisation
+---
+
+# Optimise your installation
+
+Now that you have Dendrite running, the following tweaks will improve the reliability
+and performance of your installation.
+
+## File descriptor limit
+
+Most platforms have a limit on how many file descriptors a single process can open. All
+connections made by Dendrite consume file descriptors — this includes database connections
+and network requests to remote homeservers. When participating in large federated rooms
+where Dendrite must talk to many remote servers, it is often very easy to exhaust default
+limits which are quite low.
+
+We currently recommend setting the file descriptor limit to 65535 to avoid such
+issues. Dendrite will log immediately after startup if the file descriptor limit is too low:
+
+```
+level=warning msg="IMPORTANT: Process file descriptor limit is currently 1024, it is recommended to raise the limit for Dendrite to at least 65535 to avoid issues"
+```
+
+UNIX systems have two limits: a hard limit and a soft limit. You can view the soft limit
+by running `ulimit -Sn` and the hard limit with `ulimit -Hn`:
+
+```bash
+$ ulimit -Hn
+1048576
+
+$ ulimit -Sn
+1024
+```
+
+Increase the soft limit before starting Dendrite:
+
+```bash
+ulimit -Sn 65535
+```
+
+The log line at startup should no longer appear if the limit is sufficient.
+
+If you are running under a systemd service, you can instead add `LimitNOFILE=65535` option
+to the `[Service]` section of your service unit file.
+
+## DNS caching
+
+Dendrite has a built-in DNS cache which significantly reduces the load that Dendrite will
+place on your DNS resolver. This may also speed up outbound federation.
+
+Consider enabling the DNS cache by modifying the `global` section of your configuration file:
+
+```yaml
+  dns_cache:
+    enabled: true
+    cache_size: 4096
+    cache_lifetime: 600s
+```
+
+## Time synchronisation
+
+Matrix relies heavily on TLS which requires the system time to be correct. If the clock
+drifts then you may find that federation no works reliably (or at all) and clients may
+struggle to connect to your Dendrite server.
+
+Ensure that the time is synchronised on your system by enabling NTP sync.

federationapi/consumers/presence.go

@@ -133,7 +133,7 @@ func (t *OutputPresenceConsumer) onMessage(ctx context.Context, msg *nats.Msg) b
 		return true
 	}
 
-	log.Debugf("sending presence EDU to %d servers", len(joined))
+	log.Tracef("sending presence EDU to %d servers", len(joined))
 	if err = t.queues.SendEDU(edu, t.ServerName, joined); err != nil {
 		log.WithError(err).Error("failed to send EDU")
 		return false

federationapi/federationapi.go

@@ -63,6 +63,7 @@ func AddPublicRoutes(
 		TopicSendToDeviceEvent: cfg.Matrix.JetStream.Prefixed(jetstream.OutputSendToDeviceEvent),
 		TopicTypingEvent:       cfg.Matrix.JetStream.Prefixed(jetstream.OutputTypingEvent),
 		TopicPresenceEvent:     cfg.Matrix.JetStream.Prefixed(jetstream.OutputPresenceEvent),
+		TopicDeviceListUpdate:  cfg.Matrix.JetStream.Prefixed(jetstream.InputDeviceListUpdate),
 		ServerName:             cfg.Matrix.ServerName,
 		UserAPI:                userAPI,
 	}

federationapi/producers/syncapi.go

@@ -17,6 +17,7 @@ package producers
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"strconv"
 	"time"
 
@@ -34,6 +35,7 @@ type SyncAPIProducer struct {
 	TopicSendToDeviceEvent string
 	TopicTypingEvent       string
 	TopicPresenceEvent     string
+	TopicDeviceListUpdate  string
 	JetStream              nats.JetStreamContext
 	ServerName             gomatrixserverlib.ServerName
 	UserAPI                userapi.UserInternalAPI
@@ -157,7 +159,22 @@ func (p *SyncAPIProducer) SendPresence(
 	lastActiveTS := gomatrixserverlib.AsTimestamp(time.Now().Add(-(time.Duration(lastActiveAgo) * time.Millisecond)))
 
 	m.Header.Set("last_active_ts", strconv.Itoa(int(lastActiveTS)))
-	log.Debugf("Sending presence to syncAPI: %+v", m.Header)
+	log.Tracef("Sending presence to syncAPI: %+v", m.Header)
 	_, err := p.JetStream.PublishMsg(m, nats.Context(ctx))
 	return err
 }
+
+func (p *SyncAPIProducer) SendDeviceListUpdate(
+	ctx context.Context, deviceListUpdate *gomatrixserverlib.DeviceListUpdateEvent,
+) (err error) {
+	m := nats.NewMsg(p.TopicDeviceListUpdate)
+	m.Header.Set(jetstream.UserID, deviceListUpdate.UserID)
+	m.Data, err = json.Marshal(deviceListUpdate)
+	if err != nil {
+		return fmt.Errorf("json.Marshal: %w", err)
+	}
+
+	log.Debugf("Sending device list update: %+v", m.Header)
+	_, err = p.JetStream.PublishMsg(m, nats.Context(ctx))
+	return err
+}

federationapi/routing/devices.go

@@ -85,6 +85,9 @@ func GetUserDevices(
 			if targetKey, ok := targetUser[gomatrixserverlib.KeyID(dev.DeviceID)]; ok {
 				for sourceUserID, forSourceUser := range targetKey {
 					for sourceKeyID, sourceKey := range forSourceUser {
+						if device.Keys.Signatures == nil {
+							device.Keys.Signatures = map[string]map[gomatrixserverlib.KeyID]gomatrixserverlib.Base64Bytes{}
+						}
 						if _, ok := device.Keys.Signatures[sourceUserID]; !ok {
 							device.Keys.Signatures[sourceUserID] = map[gomatrixserverlib.KeyID]gomatrixserverlib.Base64Bytes{}
 						}

federationapi/routing/send.go

@@ -501,11 +501,7 @@ func (t *txnReq) processDeviceListUpdate(ctx context.Context, e gomatrixserverli
 	} else if serverName != t.Origin {
 		return
 	}
-	var inputRes keyapi.InputDeviceListUpdateResponse
-	t.keyAPI.InputDeviceListUpdate(context.Background(), &keyapi.InputDeviceListUpdateRequest{
-		Event: payload,
-	}, &inputRes)
-	if inputRes.Error != nil {
-		util.GetLogger(ctx).WithError(inputRes.Error).WithField("user_id", payload.UserID).Error("failed to InputDeviceListUpdate")
+	if err := t.producer.SendDeviceListUpdate(ctx, &payload); err != nil {
+		util.GetLogger(ctx).WithError(err).WithField("user_id", payload.UserID).Error("failed to InputDeviceListUpdate")
 	}
 }

go.mod

@@ -2,7 +2,7 @@ module github.com/matrix-org/dendrite
 
 replace github.com/nats-io/nats-server/v2 => github.com/neilalexander/nats-server/v2 v2.8.3-0.20220513095553-73a9a246d34f
 
-replace github.com/nats-io/nats.go => github.com/neilalexander/nats.go v1.13.1-0.20220419101051-b262d9f0be1e
+replace github.com/nats-io/nats.go => github.com/neilalexander/nats.go v1.13.1-0.20220621084451-ac518c356673
 
 require (
 	github.com/Arceliar/ironwood v0.0.0-20220306165321-319147a02d98
@@ -35,7 +35,7 @@ require (
 	github.com/matrix-org/dugong v0.0.0-20210921133753-66e6b1c67e2e
 	github.com/matrix-org/go-sqlite3-js v0.0.0-20220419092513-28aa791a1c91
 	github.com/matrix-org/gomatrix v0.0.0-20210324163249-be2af5ef2e16
-	github.com/matrix-org/gomatrixserverlib v0.0.0-20220607143425-e55d796fd0b3
+	github.com/matrix-org/gomatrixserverlib v0.0.0-20220701090733-da53994b0c7f
 	github.com/matrix-org/pinecone v0.0.0-20220408153826-2999ea29ed48
 	github.com/matrix-org/util v0.0.0-20200807132607-55161520e1d4
 	github.com/matryer/is v1.4.0

go.sum

@@ -421,8 +421,8 @@ github.com/matrix-org/go-sqlite3-js v0.0.0-20220419092513-28aa791a1c91/go.mod h1
 github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26/go.mod h1:3fxX6gUjWyI/2Bt7J1OLhpCzOfO/bB3AiX0cJtEKud0=
 github.com/matrix-org/gomatrix v0.0.0-20210324163249-be2af5ef2e16 h1:ZtO5uywdd5dLDCud4r0r55eP4j9FuUNpl60Gmntcop4=
 github.com/matrix-org/gomatrix v0.0.0-20210324163249-be2af5ef2e16/go.mod h1:/gBX06Kw0exX1HrwmoBibFA98yBk/jxKpGVeyQbff+s=
-github.com/matrix-org/gomatrixserverlib v0.0.0-20220607143425-e55d796fd0b3 h1:2eYcBt8Kg+nW/xIJY5x8Uo2dQLjUF+oxLap00uFC5l8=
-github.com/matrix-org/gomatrixserverlib v0.0.0-20220607143425-e55d796fd0b3/go.mod h1:jX38yp3SSLJNftBg3PXU1ayd0PCLIiDHQ4xAc9DIixk=
+github.com/matrix-org/gomatrixserverlib v0.0.0-20220701090733-da53994b0c7f h1:XF2+J6sOq07yhK1I7ItwsgRwXorjj7gqiCvgZ4dn8W8=
+github.com/matrix-org/gomatrixserverlib v0.0.0-20220701090733-da53994b0c7f/go.mod h1:jX38yp3SSLJNftBg3PXU1ayd0PCLIiDHQ4xAc9DIixk=
 github.com/matrix-org/pinecone v0.0.0-20220408153826-2999ea29ed48 h1:W0sjjC6yjskHX4mb0nk3p0fXAlbU5bAFUFeEtlrPASE=
 github.com/matrix-org/pinecone v0.0.0-20220408153826-2999ea29ed48/go.mod h1:ulJzsVOTssIVp1j/m5eI//4VpAGDkMt5NrRuAVX7wpc=
 github.com/matrix-org/util v0.0.0-20190711121626-527ce5ddefc7/go.mod h1:vVQlW/emklohkZnOPwD3LrZUBqdfsbiyO3p1lNV8F6U=
@@ -487,8 +487,8 @@ github.com/neelance/astrewrite v0.0.0-20160511093645-99348263ae86/go.mod h1:kHJE
 github.com/neelance/sourcemap v0.0.0-20151028013722-8c68805598ab/go.mod h1:Qr6/a/Q4r9LP1IltGz7tA7iOK1WonHEYhu1HRBA7ZiM=
 github.com/neilalexander/nats-server/v2 v2.8.3-0.20220513095553-73a9a246d34f h1:Fc+TjdV1mOy0oISSzfoxNWdTqjg7tN/Vdgf+B2cwvdo=
 github.com/neilalexander/nats-server/v2 v2.8.3-0.20220513095553-73a9a246d34f/go.mod h1:vIdpKz3OG+DCg4q/xVPdXHoztEyKDWRtykQ4N7hd7C4=
-github.com/neilalexander/nats.go v1.13.1-0.20220419101051-b262d9f0be1e h1:kNIzIzj2OvnlreA+sTJ12nWJzTP3OSLNKDL/Iq9mF6Y=
-github.com/neilalexander/nats.go v1.13.1-0.20220419101051-b262d9f0be1e/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
+github.com/neilalexander/nats.go v1.13.1-0.20220621084451-ac518c356673 h1:TcKfa3Tf0qwUotv63PQVu2d1bBoLi2iEA4RHVMGDh5M=
+github.com/neilalexander/nats.go v1.13.1-0.20220621084451-ac518c356673/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
 github.com/neilalexander/utp v0.1.1-0.20210727203401-54ae7b1cd5f9 h1:lrVQzBtkeQEGGYUHwSX1XPe1E5GL6U3KYCNe2G4bncQ=
 github.com/neilalexander/utp v0.1.1-0.20210727203401-54ae7b1cd5f9/go.mod h1:NPHGhPc0/wudcaCqL/H5AOddkRf8GPRhzOujuUKGQu8=
 github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646 h1:zYyBkD/k9seD2A7fsi6Oo2LfFZAehjjQMERAvZLEDnQ=

internal/version.go

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

keyserver/api/api.go

@@ -62,8 +62,6 @@ type FederationKeyAPI interface {
 	QueryKeys(ctx context.Context, req *QueryKeysRequest, res *QueryKeysResponse)
 	QuerySignatures(ctx context.Context, req *QuerySignaturesRequest, res *QuerySignaturesResponse)
 	QueryDeviceMessages(ctx context.Context, req *QueryDeviceMessagesRequest, res *QueryDeviceMessagesResponse)
-	// InputDeviceListUpdate from a federated server EDU
-	InputDeviceListUpdate(ctx context.Context, req *InputDeviceListUpdateRequest, res *InputDeviceListUpdateResponse)
 	PerformUploadDeviceKeys(ctx context.Context, req *PerformUploadDeviceKeysRequest, res *PerformUploadDeviceKeysResponse)
 	PerformClaimKeys(ctx context.Context, req *PerformClaimKeysRequest, res *PerformClaimKeysResponse)
 }
@@ -337,11 +335,3 @@ type QuerySignaturesResponse struct {
 	// The request error, if any
 	Error *KeyError
 }
-
-type InputDeviceListUpdateRequest struct {
-	Event gomatrixserverlib.DeviceListUpdateEvent
-}
-
-type InputDeviceListUpdateResponse struct {
-	Error *KeyError
-}

keyserver/consumers/devicelistupdate.go

@@ -0,0 +1,82 @@
+// 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 consumers
+
+import (
+	"context"
+	"encoding/json"
+
+	"github.com/matrix-org/dendrite/keyserver/internal"
+	"github.com/matrix-org/dendrite/setup/config"
+	"github.com/matrix-org/dendrite/setup/jetstream"
+	"github.com/matrix-org/dendrite/setup/process"
+	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/nats-io/nats.go"
+	"github.com/sirupsen/logrus"
+)
+
+// DeviceListUpdateConsumer consumes device list updates that came in over federation.
+type DeviceListUpdateConsumer struct {
+	ctx       context.Context
+	jetstream nats.JetStreamContext
+	durable   string
+	topic     string
+	updater   *internal.DeviceListUpdater
+}
+
+// NewDeviceListUpdateConsumer creates a new DeviceListConsumer. Call Start() to begin consuming from key servers.
+func NewDeviceListUpdateConsumer(
+	process *process.ProcessContext,
+	cfg *config.KeyServer,
+	js nats.JetStreamContext,
+	updater *internal.DeviceListUpdater,
+) *DeviceListUpdateConsumer {
+	return &DeviceListUpdateConsumer{
+		ctx:       process.Context(),
+		jetstream: js,
+		durable:   cfg.Matrix.JetStream.Prefixed("KeyServerInputDeviceListConsumer"),
+		topic:     cfg.Matrix.JetStream.Prefixed(jetstream.InputDeviceListUpdate),
+		updater:   updater,
+	}
+}
+
+// Start consuming from key servers
+func (t *DeviceListUpdateConsumer) Start() error {
+	return jetstream.JetStreamConsumer(
+		t.ctx, t.jetstream, t.topic, t.durable, t.onMessage,
+		nats.DeliverAll(), nats.ManualAck(),
+	)
+}
+
+// onMessage is called in response to a message received on the
+// key change events topic from the key server.
+func (t *DeviceListUpdateConsumer) onMessage(ctx context.Context, msg *nats.Msg) bool {
+	var m gomatrixserverlib.DeviceListUpdateEvent
+	if err := json.Unmarshal(msg.Data, &m); err != nil {
+		logrus.WithError(err).Errorf("Failed to read from device list update input topic")
+		return true
+	}
+	err := t.updater.Update(ctx, m)
+	if err != nil {
+		logrus.WithFields(logrus.Fields{
+			"user_id":   m.UserID,
+			"device_id": m.DeviceID,
+			"stream_id": m.StreamID,
+			"prev_id":   m.PrevID,
+		}).WithError(err).Errorf("Failed to update device list")
+		return false
+	}
+	return true
+}

keyserver/internal/internal.go

@@ -47,17 +47,6 @@ func (a *KeyInternalAPI) SetUserAPI(i userapi.KeyserverUserAPI) {
 	a.UserAPI = i
 }
 
-func (a *KeyInternalAPI) InputDeviceListUpdate(
-	ctx context.Context, req *api.InputDeviceListUpdateRequest, res *api.InputDeviceListUpdateResponse,
-) {
-	err := a.Updater.Update(ctx, req.Event)
-	if err != nil {
-		res.Error = &api.KeyError{
-			Err: fmt.Sprintf("failed to update device list: %s", err),
-		}
-	}
-}
-
 func (a *KeyInternalAPI) QueryKeyChanges(ctx context.Context, req *api.QueryKeyChangesRequest, res *api.QueryKeyChangesResponse) {
 	userIDs, latest, err := a.DB.KeyChanges(ctx, req.Offset, req.ToOffset)
 	if err != nil {

keyserver/inthttp/client.go

@@ -63,20 +63,6 @@ type httpKeyInternalAPI struct {
 func (h *httpKeyInternalAPI) SetUserAPI(i userapi.KeyserverUserAPI) {
 	// no-op: doesn't need it
 }
-func (h *httpKeyInternalAPI) InputDeviceListUpdate(
-	ctx context.Context, req *api.InputDeviceListUpdateRequest, res *api.InputDeviceListUpdateResponse,
-) {
-	span, ctx := opentracing.StartSpanFromContext(ctx, "InputDeviceListUpdate")
-	defer span.Finish()
-
-	apiURL := h.apiURL + InputDeviceListUpdatePath
-	err := httputil.PostJSON(ctx, span, h.httpClient, apiURL, req, res)
-	if err != nil {
-		res.Error = &api.KeyError{
-			Err: err.Error(),
-		}
-	}
-}
 
 func (h *httpKeyInternalAPI) PerformClaimKeys(
 	ctx context.Context,

keyserver/inthttp/server.go

@@ -25,17 +25,6 @@ import (
 )
 
 func AddRoutes(internalAPIMux *mux.Router, s api.KeyInternalAPI) {
-	internalAPIMux.Handle(InputDeviceListUpdatePath,
-		httputil.MakeInternalAPI("inputDeviceListUpdate", func(req *http.Request) util.JSONResponse {
-			request := api.InputDeviceListUpdateRequest{}
-			response := api.InputDeviceListUpdateResponse{}
-			if err := json.NewDecoder(req.Body).Decode(&request); err != nil {
-				return util.MessageResponse(http.StatusBadRequest, err.Error())
-			}
-			s.InputDeviceListUpdate(req.Context(), &request, &response)
-			return util.JSONResponse{Code: http.StatusOK, JSON: &response}
-		}),
-	)
 	internalAPIMux.Handle(PerformClaimKeysPath,
 		httputil.MakeInternalAPI("performClaimKeys", func(req *http.Request) util.JSONResponse {
 			request := api.PerformClaimKeysRequest{}

keyserver/keyserver.go

@@ -18,6 +18,7 @@ import (
 	"github.com/gorilla/mux"
 	fedsenderapi "github.com/matrix-org/dendrite/federationapi/api"
 	"github.com/matrix-org/dendrite/keyserver/api"
+	"github.com/matrix-org/dendrite/keyserver/consumers"
 	"github.com/matrix-org/dendrite/keyserver/internal"
 	"github.com/matrix-org/dendrite/keyserver/inthttp"
 	"github.com/matrix-org/dendrite/keyserver/producers"
@@ -59,10 +60,17 @@ func NewInternalAPI(
 	updater := internal.NewDeviceListUpdater(db, ap, keyChangeProducer, fedClient, 8) // 8 workers TODO: configurable
 	ap.Updater = updater
 	go func() {
-		if err := updater.Start(); err != nil {
+		if err = updater.Start(); err != nil {
 			logrus.WithError(err).Panicf("failed to start device list updater")
 		}
 	}()
 
+	dlConsumer := consumers.NewDeviceListUpdateConsumer(
+		base.ProcessContext, cfg, js, updater,
+	)
+	if err = dlConsumer.Start(); err != nil {
+		logrus.WithError(err).Panic("failed to start device list consumer")
+	}
+
 	return ap
 }

roomserver/api/api.go

@@ -140,11 +140,8 @@ type ClientRoomserverAPI interface {
 
 	// PerformRoomUpgrade upgrades a room to a newer version
 	PerformRoomUpgrade(ctx context.Context, req *PerformRoomUpgradeRequest, resp *PerformRoomUpgradeResponse)
-	PerformAdminEvacuateRoom(
-		ctx context.Context,
-		req *PerformAdminEvacuateRoomRequest,
-		res *PerformAdminEvacuateRoomResponse,
-	)
+	PerformAdminEvacuateRoom(ctx context.Context, req *PerformAdminEvacuateRoomRequest, res *PerformAdminEvacuateRoomResponse)
+	PerformAdminEvacuateUser(ctx context.Context, req *PerformAdminEvacuateUserRequest, res *PerformAdminEvacuateUserResponse)
 	PerformPeek(ctx context.Context, req *PerformPeekRequest, res *PerformPeekResponse)
 	PerformUnpeek(ctx context.Context, req *PerformUnpeekRequest, res *PerformUnpeekResponse)
 	PerformInvite(ctx context.Context, req *PerformInviteRequest, res *PerformInviteResponse) error
@@ -161,6 +158,7 @@ type UserRoomserverAPI interface {
 	QueryLatestEventsAndStateAPI
 	QueryCurrentState(ctx context.Context, req *QueryCurrentStateRequest, res *QueryCurrentStateResponse) error
 	QueryMembershipsForRoom(ctx context.Context, req *QueryMembershipsForRoomRequest, res *QueryMembershipsForRoomResponse) error
+	PerformAdminEvacuateUser(ctx context.Context, req *PerformAdminEvacuateUserRequest, res *PerformAdminEvacuateUserResponse)
 }
 
 type FederationRoomserverAPI interface {

roomserver/api/api_trace.go

@@ -113,6 +113,15 @@ func (t *RoomserverInternalAPITrace) PerformAdminEvacuateRoom(
 	util.GetLogger(ctx).Infof("PerformAdminEvacuateRoom req=%+v res=%+v", js(req), js(res))
 }
 
+func (t *RoomserverInternalAPITrace) PerformAdminEvacuateUser(
+	ctx context.Context,
+	req *PerformAdminEvacuateUserRequest,
+	res *PerformAdminEvacuateUserResponse,
+) {
+	t.Impl.PerformAdminEvacuateUser(ctx, req, res)
+	util.GetLogger(ctx).Infof("PerformAdminEvacuateUser req=%+v res=%+v", js(req), js(res))
+}
+
 func (t *RoomserverInternalAPITrace) PerformInboundPeek(
 	ctx context.Context,
 	req *PerformInboundPeekRequest,

roomserver/api/output.go

@@ -161,6 +161,8 @@ type OutputNewRoomEvent struct {
 	// The transaction ID of the send request if sent by a local user and one
 	// was specified
 	TransactionID *TransactionID `json:"transaction_id,omitempty"`
+	// The history visibility of the event.
+	HistoryVisibility gomatrixserverlib.HistoryVisibility `json:"history_visibility"`
 }
 
 func (o *OutputNewRoomEvent) NeededStateEventIDs() ([]*gomatrixserverlib.HeaderedEvent, []string) {
@@ -187,7 +189,8 @@ func (o *OutputNewRoomEvent) NeededStateEventIDs() ([]*gomatrixserverlib.Headere
 // should build their current room state up from OutputNewRoomEvents only.
 type OutputOldRoomEvent struct {
 	// The Event.
-	Event *gomatrixserverlib.HeaderedEvent `json:"event"`
+	Event             *gomatrixserverlib.HeaderedEvent    `json:"event"`
+	HistoryVisibility gomatrixserverlib.HistoryVisibility `json:"history_visibility"`
 }
 
 // An OutputNewInviteEvent is written whenever an invite becomes active.

roomserver/api/perform.go

@@ -223,3 +223,12 @@ type PerformAdminEvacuateRoomResponse struct {
 	Affected []string `json:"affected"`
 	Error    *PerformError
 }
+
+type PerformAdminEvacuateUserRequest struct {
+	UserID string `json:"user_id"`
+}
+
+type PerformAdminEvacuateUserResponse struct {
+	Affected []string `json:"affected"`
+	Error    *PerformError
+}

roomserver/internal/alias.go

@@ -216,11 +216,10 @@ func (r *RoomserverInternalAPI) RemoveRoomAlias(
 				return err
 			}
 
-			err = api.SendEvents(ctx, r.RSAPI, api.KindNew, []*gomatrixserverlib.HeaderedEvent{newEvent}, r.ServerName, r.ServerName, nil, false)
+			err = api.SendEvents(ctx, r, api.KindNew, []*gomatrixserverlib.HeaderedEvent{newEvent}, r.ServerName, r.ServerName, nil, false)
 			if err != nil {
 				return err
 			}
-
 		}
 	}
 

roomserver/internal/api.go

@@ -12,8 +12,10 @@ import (
 	"github.com/matrix-org/dendrite/roomserver/internal/input"
 	"github.com/matrix-org/dendrite/roomserver/internal/perform"
 	"github.com/matrix-org/dendrite/roomserver/internal/query"
+	"github.com/matrix-org/dendrite/roomserver/producers"
 	"github.com/matrix-org/dendrite/roomserver/storage"
 	"github.com/matrix-org/dendrite/setup/config"
+	"github.com/matrix-org/dendrite/setup/jetstream"
 	"github.com/matrix-org/dendrite/setup/process"
 	userapi "github.com/matrix-org/dendrite/userapi/api"
 	"github.com/matrix-org/gomatrixserverlib"
@@ -49,17 +51,21 @@ type RoomserverInternalAPI struct {
 	JetStream              nats.JetStreamContext
 	Durable                string
 	InputRoomEventTopic    string // JetStream topic for new input room events
-	OutputRoomEventTopic   string // JetStream topic for new output room events
+	OutputProducer         *producers.RoomEventProducer
 	PerspectiveServerNames []gomatrixserverlib.ServerName
 }
 
 func NewRoomserverAPI(
 	processCtx *process.ProcessContext, cfg *config.RoomServer, roomserverDB storage.Database,
-	consumer nats.JetStreamContext, nc *nats.Conn,
-	inputRoomEventTopic, outputRoomEventTopic string,
+	js nats.JetStreamContext, nc *nats.Conn, inputRoomEventTopic string,
 	caches caching.RoomServerCaches, perspectiveServerNames []gomatrixserverlib.ServerName,
 ) *RoomserverInternalAPI {
 	serverACLs := acls.NewServerACLs(roomserverDB)
+	producer := &producers.RoomEventProducer{
+		Topic:     string(cfg.Matrix.JetStream.Prefixed(jetstream.OutputRoomEvent)),
+		JetStream: js,
+		ACLs:      serverACLs,
+	}
 	a := &RoomserverInternalAPI{
 		ProcessContext:         processCtx,
 		DB:                     roomserverDB,
@@ -68,8 +74,8 @@ func NewRoomserverAPI(
 		ServerName:             cfg.Matrix.ServerName,
 		PerspectiveServerNames: perspectiveServerNames,
 		InputRoomEventTopic:    inputRoomEventTopic,
-		OutputRoomEventTopic:   outputRoomEventTopic,
-		JetStream:              consumer,
+		OutputProducer:         producer,
+		JetStream:              js,
 		NATSClient:             nc,
 		Durable:                cfg.Matrix.JetStream.Durable("RoomserverInputConsumer"),
 		ServerACLs:             serverACLs,
@@ -92,19 +98,19 @@ func (r *RoomserverInternalAPI) SetFederationAPI(fsAPI fsAPI.RoomserverFederatio
 	r.KeyRing = keyRing
 
 	r.Inputer = &input.Inputer{
-		Cfg:                  r.Cfg,
-		ProcessContext:       r.ProcessContext,
-		DB:                   r.DB,
-		InputRoomEventTopic:  r.InputRoomEventTopic,
-		OutputRoomEventTopic: r.OutputRoomEventTopic,
-		JetStream:            r.JetStream,
-		NATSClient:           r.NATSClient,
-		Durable:              nats.Durable(r.Durable),
-		ServerName:           r.Cfg.Matrix.ServerName,
-		FSAPI:                fsAPI,
-		KeyRing:              keyRing,
-		ACLs:                 r.ServerACLs,
-		Queryer:              r.Queryer,
+		Cfg:                 r.Cfg,
+		ProcessContext:      r.ProcessContext,
+		DB:                  r.DB,
+		InputRoomEventTopic: r.InputRoomEventTopic,
+		OutputProducer:      r.OutputProducer,
+		JetStream:           r.JetStream,
+		NATSClient:          r.NATSClient,
+		Durable:             nats.Durable(r.Durable),
+		ServerName:          r.Cfg.Matrix.ServerName,
+		FSAPI:               fsAPI,
+		KeyRing:             keyRing,
+		ACLs:                r.ServerACLs,
+		Queryer:             r.Queryer,
 	}
 	r.Inviter = &perform.Inviter{
 		DB:      r.DB,
@@ -170,6 +176,7 @@ func (r *RoomserverInternalAPI) SetFederationAPI(fsAPI fsAPI.RoomserverFederatio
 		Cfg:     r.Cfg,
 		Inputer: r.Inputer,
 		Queryer: r.Queryer,
+		Leaver:  r.Leaver,
 	}
 
 	if err := r.Inputer.Start(); err != nil {
@@ -198,7 +205,7 @@ func (r *RoomserverInternalAPI) PerformInvite(
 	if len(outputEvents) == 0 {
 		return nil
 	}
-	return r.WriteOutputEvents(req.Event.RoomID(), outputEvents)
+	return r.OutputProducer.ProduceRoomEvents(req.Event.RoomID(), outputEvents)
 }
 
 func (r *RoomserverInternalAPI) PerformLeave(
@@ -214,7 +221,7 @@ func (r *RoomserverInternalAPI) PerformLeave(
 	if len(outputEvents) == 0 {
 		return nil
 	}
-	return r.WriteOutputEvents(req.RoomID, outputEvents)
+	return r.OutputProducer.ProduceRoomEvents(req.RoomID, outputEvents)
 }
 
 func (r *RoomserverInternalAPI) PerformForget(

roomserver/internal/input/input.go

@@ -29,6 +29,7 @@ import (
 	"github.com/matrix-org/dendrite/roomserver/acls"
 	"github.com/matrix-org/dendrite/roomserver/api"
 	"github.com/matrix-org/dendrite/roomserver/internal/query"
+	"github.com/matrix-org/dendrite/roomserver/producers"
 	"github.com/matrix-org/dendrite/roomserver/storage"
 	"github.com/matrix-org/dendrite/setup/config"
 	"github.com/matrix-org/dendrite/setup/jetstream"
@@ -37,16 +38,8 @@ import (
 	"github.com/nats-io/nats.go"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/sirupsen/logrus"
-	log "github.com/sirupsen/logrus"
-	"github.com/tidwall/gjson"
 )
 
-var keyContentFields = map[string]string{
-	"m.room.join_rules":         "join_rule",
-	"m.room.history_visibility": "history_visibility",
-	"m.room.member":             "membership",
-}
-
 // Inputer is responsible for consuming from the roomserver input
 // streams and processing the events. All input events are queued
 // into a single NATS stream and the order is preserved strictly.
@@ -75,19 +68,19 @@ var keyContentFields = map[string]string{
 // up, so they will do nothing until a new event comes in for B
 // or C.
 type Inputer struct {
-	Cfg                  *config.RoomServer
-	ProcessContext       *process.ProcessContext
-	DB                   storage.Database
-	NATSClient           *nats.Conn
-	JetStream            nats.JetStreamContext
-	Durable              nats.SubOpt
-	ServerName           gomatrixserverlib.ServerName
-	FSAPI                fedapi.RoomserverFederationAPI
-	KeyRing              gomatrixserverlib.JSONVerifier
-	ACLs                 *acls.ServerACLs
-	InputRoomEventTopic  string
-	OutputRoomEventTopic string
-	workers              sync.Map // room ID -> *worker
+	Cfg                 *config.RoomServer
+	ProcessContext      *process.ProcessContext
+	DB                  storage.Database
+	NATSClient          *nats.Conn
+	JetStream           nats.JetStreamContext
+	Durable             nats.SubOpt
+	ServerName          gomatrixserverlib.ServerName
+	FSAPI               fedapi.RoomserverFederationAPI
+	KeyRing             gomatrixserverlib.JSONVerifier
+	ACLs                *acls.ServerACLs
+	InputRoomEventTopic string
+	OutputProducer      *producers.RoomEventProducer
+	workers             sync.Map // room ID -> *worker
 
 	Queryer *query.Queryer
 }
@@ -370,58 +363,6 @@ func (r *Inputer) InputRoomEvents(
 	}
 }
 
-// WriteOutputEvents implements OutputRoomEventWriter
-func (r *Inputer) WriteOutputEvents(roomID string, updates []api.OutputEvent) error {
-	var err error
-	for _, update := range updates {
-		msg := &nats.Msg{
-			Subject: r.OutputRoomEventTopic,
-			Header:  nats.Header{},
-		}
-		msg.Header.Set(jetstream.RoomID, roomID)
-		msg.Data, err = json.Marshal(update)
-		if err != nil {
-			return err
-		}
-		logger := log.WithFields(log.Fields{
-			"room_id": roomID,
-			"type":    update.Type,
-		})
-		if update.NewRoomEvent != nil {
-			eventType := update.NewRoomEvent.Event.Type()
-			logger = logger.WithFields(log.Fields{
-				"event_type":     eventType,
-				"event_id":       update.NewRoomEvent.Event.EventID(),
-				"adds_state":     len(update.NewRoomEvent.AddsStateEventIDs),
-				"removes_state":  len(update.NewRoomEvent.RemovesStateEventIDs),
-				"send_as_server": update.NewRoomEvent.SendAsServer,
-				"sender":         update.NewRoomEvent.Event.Sender(),
-			})
-			if update.NewRoomEvent.Event.StateKey() != nil {
-				logger = logger.WithField("state_key", *update.NewRoomEvent.Event.StateKey())
-			}
-			contentKey := keyContentFields[eventType]
-			if contentKey != "" {
-				value := gjson.GetBytes(update.NewRoomEvent.Event.Content(), contentKey)
-				if value.Exists() {
-					logger = logger.WithField("content_value", value.String())
-				}
-			}
-
-			if eventType == "m.room.server_acl" && update.NewRoomEvent.Event.StateKeyEquals("") {
-				ev := update.NewRoomEvent.Event.Unwrap()
-				defer r.ACLs.OnServerACLUpdate(ev)
-			}
-		}
-		logger.Tracef("Producing to topic '%s'", r.OutputRoomEventTopic)
-		if _, err := r.JetStream.PublishMsg(msg); err != nil {
-			logger.WithError(err).Errorf("Failed to produce to topic '%s': %s", r.OutputRoomEventTopic, err)
-			return err
-		}
-	}
-	return nil
-}
-
 var roomserverInputBackpressure = prometheus.NewGaugeVec(
 	prometheus.GaugeOpts{
 		Namespace: "dendrite",

roomserver/internal/input/input_events.go

@@ -295,6 +295,22 @@ func (r *Inputer) processRoomEvent(
 		}
 	}
 
+	// Get the state before the event so that we can work out if the event was
+	// allowed at the time, and also to get the history visibility. We won't
+	// bother doing this if the event was already rejected as it just ends up
+	// burning CPU time.
+	historyVisibility := gomatrixserverlib.HistoryVisibilityJoined // Default to restrictive.
+	if rejectionErr == nil && !isRejected && !softfail {
+		var err error
+		historyVisibility, rejectionErr, err = r.processStateBefore(ctx, input, missingPrev)
+		if err != nil {
+			return fmt.Errorf("r.processStateBefore: %w", err)
+		}
+		if rejectionErr != nil {
+			isRejected = true
+		}
+	}
+
 	// Store the event.
 	_, _, stateAtEvent, redactionEvent, redactedEventID, err := r.DB.StoreEvent(ctx, event, authEventNIDs, isRejected || softfail)
 	if err != nil {
@@ -360,15 +376,17 @@ func (r *Inputer) processRoomEvent(
 			input.SendAsServer,  // send as server
 			input.TransactionID, // transaction ID
 			input.HasState,      // rewrites state?
+			historyVisibility,   // the history visibility before the event
 		); err != nil {
 			return fmt.Errorf("r.updateLatestEvents: %w", err)
 		}
 	case api.KindOld:
-		err = r.WriteOutputEvents(event.RoomID(), []api.OutputEvent{
+		err = r.OutputProducer.ProduceRoomEvents(event.RoomID(), []api.OutputEvent{
 			{
 				Type: api.OutputTypeOldRoomEvent,
 				OldRoomEvent: &api.OutputOldRoomEvent{
-					Event: headered,
+					Event:             headered,
+					HistoryVisibility: historyVisibility,
 				},
 			},
 		})
@@ -382,7 +400,7 @@ func (r *Inputer) processRoomEvent(
 	// so notify downstream components to redact this event - they should have it if they've
 	// been tracking our output log.
 	if redactedEventID != "" {
-		err = r.WriteOutputEvents(event.RoomID(), []api.OutputEvent{
+		err = r.OutputProducer.ProduceRoomEvents(event.RoomID(), []api.OutputEvent{
 			{
 				Type: api.OutputTypeRedactedEvent,
 				RedactedEvent: &api.OutputRedactedEvent{
@@ -402,6 +420,100 @@ func (r *Inputer) processRoomEvent(
 	return nil
 }
 
+// processStateBefore works out what the state is before the event and
+// then checks the event auths against the state at the time. It also
+// tries to determine what the history visibility was of the event at
+// the time, so that it can be sent in the output event to downstream
+// components.
+// nolint:nakedret
+func (r *Inputer) processStateBefore(
+	ctx context.Context,
+	input *api.InputRoomEvent,
+	missingPrev bool,
+) (historyVisibility gomatrixserverlib.HistoryVisibility, rejectionErr error, err error) {
+	historyVisibility = gomatrixserverlib.HistoryVisibilityJoined // Default to restrictive.
+	event := input.Event.Unwrap()
+	isCreateEvent := event.Type() == gomatrixserverlib.MRoomCreate && event.StateKeyEquals("")
+	var stateBeforeEvent []*gomatrixserverlib.Event
+	switch {
+	case isCreateEvent:
+		// There's no state before a create event so there is nothing
+		// else to do.
+		return
+	case input.HasState:
+		// If we're overriding the state then we need to go and retrieve
+		// them from the database. It's a hard error if they are missing.
+		stateEvents, err := r.DB.EventsFromIDs(ctx, input.StateEventIDs)
+		if err != nil {
+			return "", nil, fmt.Errorf("r.DB.EventsFromIDs: %w", err)
+		}
+		stateBeforeEvent = make([]*gomatrixserverlib.Event, 0, len(stateEvents))
+		for _, entry := range stateEvents {
+			stateBeforeEvent = append(stateBeforeEvent, entry.Event)
+		}
+	case missingPrev:
+		// We don't know all of the prev events, so we can't work out
+		// the state before the event. Reject it in that case.
+		rejectionErr = fmt.Errorf("event %q has missing prev events", event.EventID())
+		return
+	case len(event.PrevEventIDs()) == 0:
+		// There should be prev events since it's not a create event.
+		// A non-create event that claims to have no prev events is
+		// invalid, so reject it.
+		rejectionErr = fmt.Errorf("event %q must have prev events", event.EventID())
+		return
+	default:
+		// For all non-create events, there must be prev events, so we'll
+		// ask the query API for the relevant tuples needed for auth. We
+		// will include the history visibility here even though we don't
+		// actually need it for auth, because we want to send it in the
+		// output events.
+		tuplesNeeded := gomatrixserverlib.StateNeededForAuth([]*gomatrixserverlib.Event{event}).Tuples()
+		tuplesNeeded = append(tuplesNeeded, gomatrixserverlib.StateKeyTuple{
+			EventType: gomatrixserverlib.MRoomHistoryVisibility,
+			StateKey:  "",
+		})
+		stateBeforeReq := &api.QueryStateAfterEventsRequest{
+			RoomID:       event.RoomID(),
+			PrevEventIDs: event.PrevEventIDs(),
+			StateToFetch: tuplesNeeded,
+		}
+		stateBeforeRes := &api.QueryStateAfterEventsResponse{}
+		if err := r.Queryer.QueryStateAfterEvents(ctx, stateBeforeReq, stateBeforeRes); err != nil {
+			return "", nil, fmt.Errorf("r.Queryer.QueryStateAfterEvents: %w", err)
+		}
+		switch {
+		case !stateBeforeRes.RoomExists:
+			rejectionErr = fmt.Errorf("room %q does not exist", event.RoomID())
+			return
+		case !stateBeforeRes.PrevEventsExist:
+			rejectionErr = fmt.Errorf("prev events of %q are not known", event.EventID())
+			return
+		default:
+			stateBeforeEvent = gomatrixserverlib.UnwrapEventHeaders(stateBeforeRes.StateEvents)
+		}
+	}
+	// At this point, stateBeforeEvent should be populated either by
+	// the supplied state in the input request, or from the prev events.
+	// Check whether the event is allowed or not.
+	stateBeforeAuth := gomatrixserverlib.NewAuthEvents(stateBeforeEvent)
+	if rejectionErr = gomatrixserverlib.Allowed(event, &stateBeforeAuth); rejectionErr != nil {
+		return
+	}
+	// Work out what the history visibility was at the time of the
+	// event.
+	for _, event := range stateBeforeEvent {
+		if event.Type() != gomatrixserverlib.MRoomHistoryVisibility || !event.StateKeyEquals("") {
+			continue
+		}
+		if hisVis, err := event.HistoryVisibility(); err == nil {
+			historyVisibility = hisVis
+			break
+		}
+	}
+	return
+}
+
 // fetchAuthEvents will check to see if any of the
 // auth events specified by the given event are unknown. If they are
 // then we will go off and request them from the federation and then

roomserver/internal/input/input_latest_events.go

@@ -56,6 +56,7 @@ func (r *Inputer) updateLatestEvents(
 	sendAsServer string,
 	transactionID *api.TransactionID,
 	rewritesState bool,
+	historyVisibility gomatrixserverlib.HistoryVisibility,
 ) (err error) {
 	span, ctx := opentracing.StartSpanFromContext(ctx, "updateLatestEvents")
 	defer span.Finish()
@@ -69,15 +70,16 @@ func (r *Inputer) updateLatestEvents(
 	defer sqlutil.EndTransactionWithCheck(updater, &succeeded, &err)
 
 	u := latestEventsUpdater{
-		ctx:           ctx,
-		api:           r,
-		updater:       updater,
-		roomInfo:      roomInfo,
-		stateAtEvent:  stateAtEvent,
-		event:         event,
-		sendAsServer:  sendAsServer,
-		transactionID: transactionID,
-		rewritesState: rewritesState,
+		ctx:               ctx,
+		api:               r,
+		updater:           updater,
+		roomInfo:          roomInfo,
+		stateAtEvent:      stateAtEvent,
+		event:             event,
+		sendAsServer:      sendAsServer,
+		transactionID:     transactionID,
+		rewritesState:     rewritesState,
+		historyVisibility: historyVisibility,
 	}
 
 	if err = u.doUpdateLatestEvents(); err != nil {
@@ -119,6 +121,8 @@ type latestEventsUpdater struct {
 	// The snapshots of current state before and after processing this event
 	oldStateNID types.StateSnapshotNID
 	newStateNID types.StateSnapshotNID
+	// The history visibility of the event itself (from the state before the event).
+	historyVisibility gomatrixserverlib.HistoryVisibility
 }
 
 func (u *latestEventsUpdater) doUpdateLatestEvents() error {
@@ -188,7 +192,7 @@ func (u *latestEventsUpdater) doUpdateLatestEvents() error {
 	// send the event asynchronously but we would need to ensure that 1) the events are written to the log in
 	// the correct order, 2) that pending writes are resent across restarts. In order to avoid writing all the
 	// necessary bookkeeping we'll keep the event sending synchronous for now.
-	if err = u.api.WriteOutputEvents(u.event.RoomID(), updates); err != nil {
+	if err = u.api.OutputProducer.ProduceRoomEvents(u.event.RoomID(), updates); err != nil {
 		return fmt.Errorf("u.api.WriteOutputEvents: %w", err)
 	}
 
@@ -365,12 +369,13 @@ func (u *latestEventsUpdater) makeOutputNewRoomEvent() (*api.OutputEvent, error)
 	}
 
 	ore := api.OutputNewRoomEvent{
-		Event:           u.event.Headered(u.roomInfo.RoomVersion),
-		RewritesState:   u.rewritesState,
-		LastSentEventID: u.lastEventIDSent,
-		LatestEventIDs:  latestEventIDs,
-		TransactionID:   u.transactionID,
-		SendAsServer:    u.sendAsServer,
+		Event:             u.event.Headered(u.roomInfo.RoomVersion),
+		RewritesState:     u.rewritesState,
+		LastSentEventID:   u.lastEventIDSent,
+		LatestEventIDs:    latestEventIDs,
+		TransactionID:     u.transactionID,
+		SendAsServer:      u.sendAsServer,
+		HistoryVisibility: u.historyVisibility,
 	}
 
 	eventIDMap, err := u.stateEventMap()

roomserver/internal/perform/perform_admin.go

@@ -16,6 +16,7 @@ package perform
 
 import (
 	"context"
+	"database/sql"
 	"encoding/json"
 	"fmt"
 	"time"
@@ -34,6 +35,7 @@ type Admin struct {
 	Cfg     *config.RoomServer
 	Queryer *query.Queryer
 	Inputer *input.Inputer
+	Leaver  *Leaver
 }
 
 // PerformEvacuateRoom will remove all local users from the given room.
@@ -160,3 +162,71 @@ func (r *Admin) PerformAdminEvacuateRoom(
 	inputRes := &api.InputRoomEventsResponse{}
 	r.Inputer.InputRoomEvents(ctx, inputReq, inputRes)
 }
+
+func (r *Admin) PerformAdminEvacuateUser(
+	ctx context.Context,
+	req *api.PerformAdminEvacuateUserRequest,
+	res *api.PerformAdminEvacuateUserResponse,
+) {
+	_, domain, err := gomatrixserverlib.SplitID('@', req.UserID)
+	if err != nil {
+		res.Error = &api.PerformError{
+			Code: api.PerformErrorBadRequest,
+			Msg:  fmt.Sprintf("Malformed user ID: %s", err),
+		}
+		return
+	}
+	if domain != r.Cfg.Matrix.ServerName {
+		res.Error = &api.PerformError{
+			Code: api.PerformErrorBadRequest,
+			Msg:  "Can only evacuate local users using this endpoint",
+		}
+		return
+	}
+
+	roomIDs, err := r.DB.GetRoomsByMembership(ctx, req.UserID, gomatrixserverlib.Join)
+	if err != nil && err != sql.ErrNoRows {
+		res.Error = &api.PerformError{
+			Code: api.PerformErrorBadRequest,
+			Msg:  fmt.Sprintf("r.DB.GetRoomsByMembership: %s", err),
+		}
+		return
+	}
+
+	inviteRoomIDs, err := r.DB.GetRoomsByMembership(ctx, req.UserID, gomatrixserverlib.Invite)
+	if err != nil && err != sql.ErrNoRows {
+		res.Error = &api.PerformError{
+			Code: api.PerformErrorBadRequest,
+			Msg:  fmt.Sprintf("r.DB.GetRoomsByMembership: %s", err),
+		}
+		return
+	}
+
+	for _, roomID := range append(roomIDs, inviteRoomIDs...) {
+		leaveReq := &api.PerformLeaveRequest{
+			RoomID: roomID,
+			UserID: req.UserID,
+		}
+		leaveRes := &api.PerformLeaveResponse{}
+		outputEvents, err := r.Leaver.PerformLeave(ctx, leaveReq, leaveRes)
+		if err != nil {
+			res.Error = &api.PerformError{
+				Code: api.PerformErrorBadRequest,
+				Msg:  fmt.Sprintf("r.Leaver.PerformLeave: %s", err),
+			}
+			return
+		}
+		if len(outputEvents) == 0 {
+			continue
+		}
+		if err := r.Inputer.OutputProducer.ProduceRoomEvents(roomID, outputEvents); err != nil {
+			res.Error = &api.PerformError{
+				Code: api.PerformErrorBadRequest,
+				Msg:  fmt.Sprintf("r.Inputer.WriteOutputEvents: %s", err),
+			}
+			return
+		}
+
+		res.Affected = append(res.Affected, roomID)
+	}
+}

roomserver/internal/perform/perform_backfill.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/getsentry/sentry-go"
 	federationAPI "github.com/matrix-org/dendrite/federationapi/api"
 	"github.com/matrix-org/dendrite/internal/eventutil"
 	"github.com/matrix-org/dendrite/roomserver/api"
@@ -206,8 +207,17 @@ func (r *Backfiller) fetchAndStoreMissingEvents(ctx context.Context, roomVer gom
 			}
 			logger.Infof("returned %d PDUs which made events %+v", len(res.PDUs), result)
 			for _, res := range result {
-				if res.Error != nil {
-					logger.WithError(res.Error).Warn("event failed PDU checks")
+				switch err := res.Error.(type) {
+				case nil:
+				case gomatrixserverlib.SignatureErr:
+					// The signature of the event might not be valid anymore, for example if
+					// the key ID was reused with a different signature.
+					logger.WithError(err).Errorf("event failed PDU checks, storing anyway")
+				case gomatrixserverlib.AuthChainErr, gomatrixserverlib.AuthRulesErr:
+					logger.WithError(err).Warn("event failed PDU checks")
+					continue
+				default:
+					logger.WithError(err).Warn("event failed PDU checks")
 					continue
 				}
 				missingMap[id] = res.Event
@@ -306,6 +316,7 @@ FederationHit:
 		b.eventIDToBeforeStateIDs[targetEvent.EventID()] = res
 		return res, nil
 	}
+	sentry.CaptureException(lastErr) // temporary to see if we might need to raise the server limit
 	return nil, lastErr
 }
 
@@ -366,19 +377,25 @@ func (b *backfillRequester) StateBeforeEvent(ctx context.Context, roomVer gomatr
 		}
 	}
 
-	c := gomatrixserverlib.FederatedStateProvider{
-		FedClient:          b.fsAPI,
-		RememberAuthEvents: false,
-		Server:             b.servers[0],
+	var lastErr error
+	for _, srv := range b.servers {
+		c := gomatrixserverlib.FederatedStateProvider{
+			FedClient:          b.fsAPI,
+			RememberAuthEvents: false,
+			Server:             srv,
+		}
+		result, err := c.StateBeforeEvent(ctx, roomVer, event, eventIDs)
+		if err != nil {
+			lastErr = err
+			continue
+		}
+		for eventID, ev := range result {
+			b.eventIDMap[eventID] = ev
+		}
+		return result, nil
 	}
-	result, err := c.StateBeforeEvent(ctx, roomVer, event, eventIDs)
-	if err != nil {
-		return nil, err
-	}
-	for eventID, ev := range result {
-		b.eventIDMap[eventID] = ev
-	}
-	return result, nil
+	sentry.CaptureException(lastErr) // temporary to see if we might need to raise the server limit
+	return nil, lastErr
 }
 
 // ServersAtEvent is called when trying to determine which server to request from.

roomserver/internal/perform/perform_inbound_peek.go

@@ -113,7 +113,7 @@ func (r *InboundPeeker) PerformInboundPeek(
 		response.AuthChainEvents = append(response.AuthChainEvents, event.Headered(info.RoomVersion))
 	}
 
-	err = r.Inputer.WriteOutputEvents(request.RoomID, []api.OutputEvent{
+	err = r.Inputer.OutputProducer.ProduceRoomEvents(request.RoomID, []api.OutputEvent{
 		{
 			Type: api.OutputTypeNewInboundPeek,
 			NewInboundPeek: &api.OutputNewInboundPeek{

roomserver/internal/perform/perform_invite.go

@@ -56,7 +56,14 @@ func (r *Inviter) PerformInvite(
 		return nil, fmt.Errorf("failed to load RoomInfo: %w", err)
 	}
 
-	_, domain, _ := gomatrixserverlib.SplitID('@', targetUserID)
+	_, domain, err := gomatrixserverlib.SplitID('@', targetUserID)
+	if err != nil {
+		res.Error = &api.PerformError{
+			Code: api.PerformErrorBadRequest,
+			Msg:  fmt.Sprintf("The user ID %q is invalid!", targetUserID),
+		}
+		return nil, nil
+	}
 	isTargetLocal := domain == r.Cfg.Matrix.ServerName
 	isOriginLocal := event.Origin() == r.Cfg.Matrix.ServerName
 

roomserver/internal/perform/perform_peek.go

@@ -207,7 +207,7 @@ func (r *Peeker) performPeekRoomByID(
 
 	// TODO: handle federated peeks
 
-	err = r.Inputer.WriteOutputEvents(roomID, []api.OutputEvent{
+	err = r.Inputer.OutputProducer.ProduceRoomEvents(roomID, []api.OutputEvent{
 		{
 			Type: api.OutputTypeNewPeek,
 			NewPeek: &api.OutputNewPeek{

roomserver/internal/perform/perform_unpeek.go

@@ -96,7 +96,7 @@ func (r *Unpeeker) performUnpeekRoomByID(
 
 	// TODO: handle federated peeks
 
-	err = r.Inputer.WriteOutputEvents(req.RoomID, []api.OutputEvent{
+	err = r.Inputer.OutputProducer.ProduceRoomEvents(req.RoomID, []api.OutputEvent{
 		{
 			Type: api.OutputTypeRetirePeek,
 			RetirePeek: &api.OutputRetirePeek{

roomserver/inthttp/client.go

@@ -40,6 +40,7 @@ const (
 	RoomserverPerformInboundPeekPath       = "/roomserver/performInboundPeek"
 	RoomserverPerformForgetPath            = "/roomserver/performForget"
 	RoomserverPerformAdminEvacuateRoomPath = "/roomserver/performAdminEvacuateRoom"
+	RoomserverPerformAdminEvacuateUserPath = "/roomserver/performAdminEvacuateUser"
 
 	// Query operations
 	RoomserverQueryLatestEventsAndStatePath    = "/roomserver/queryLatestEventsAndState"
@@ -305,6 +306,23 @@ func (h *httpRoomserverInternalAPI) PerformAdminEvacuateRoom(
 	}
 }
 
+func (h *httpRoomserverInternalAPI) PerformAdminEvacuateUser(
+	ctx context.Context,
+	req *api.PerformAdminEvacuateUserRequest,
+	res *api.PerformAdminEvacuateUserResponse,
+) {
+	span, ctx := opentracing.StartSpanFromContext(ctx, "PerformAdminEvacuateUser")
+	defer span.Finish()
+
+	apiURL := h.roomserverURL + RoomserverPerformAdminEvacuateUserPath
+	err := httputil.PostJSON(ctx, span, h.httpClient, apiURL, req, res)
+	if err != nil {
+		res.Error = &api.PerformError{
+			Msg: fmt.Sprintf("failed to communicate with roomserver: %s", err),
+		}
+	}
+}
+
 // QueryLatestEventsAndState implements RoomserverQueryAPI
 func (h *httpRoomserverInternalAPI) QueryLatestEventsAndState(
 	ctx context.Context,

roomserver/inthttp/server.go

@@ -129,6 +129,17 @@ func AddRoutes(r api.RoomserverInternalAPI, internalAPIMux *mux.Router) {
 			return util.JSONResponse{Code: http.StatusOK, JSON: &response}
 		}),
 	)
+	internalAPIMux.Handle(RoomserverPerformAdminEvacuateUserPath,
+		httputil.MakeInternalAPI("performAdminEvacuateUser", func(req *http.Request) util.JSONResponse {
+			var request api.PerformAdminEvacuateUserRequest
+			var response api.PerformAdminEvacuateUserResponse
+			if err := json.NewDecoder(req.Body).Decode(&request); err != nil {
+				return util.MessageResponse(http.StatusBadRequest, err.Error())
+			}
+			r.PerformAdminEvacuateUser(req.Context(), &request, &response)
+			return util.JSONResponse{Code: http.StatusOK, JSON: &response}
+		}),
+	)
 	internalAPIMux.Handle(
 		RoomserverQueryPublishedRoomsPath,
 		httputil.MakeInternalAPI("queryPublishedRooms", func(req *http.Request) util.JSONResponse {

roomserver/producers/roomevent.go

@@ -0,0 +1,89 @@
+// 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 producers
+
+import (
+	"encoding/json"
+
+	"github.com/matrix-org/dendrite/roomserver/acls"
+	"github.com/matrix-org/dendrite/roomserver/api"
+	"github.com/matrix-org/dendrite/setup/jetstream"
+	"github.com/nats-io/nats.go"
+	log "github.com/sirupsen/logrus"
+	"github.com/tidwall/gjson"
+)
+
+var keyContentFields = map[string]string{
+	"m.room.join_rules":         "join_rule",
+	"m.room.history_visibility": "history_visibility",
+	"m.room.member":             "membership",
+}
+
+type RoomEventProducer struct {
+	Topic     string
+	ACLs      *acls.ServerACLs
+	JetStream nats.JetStreamContext
+}
+
+func (r *RoomEventProducer) ProduceRoomEvents(roomID string, updates []api.OutputEvent) error {
+	var err error
+	for _, update := range updates {
+		msg := &nats.Msg{
+			Subject: r.Topic,
+			Header:  nats.Header{},
+		}
+		msg.Header.Set(jetstream.RoomID, roomID)
+		msg.Data, err = json.Marshal(update)
+		if err != nil {
+			return err
+		}
+		logger := log.WithFields(log.Fields{
+			"room_id": roomID,
+			"type":    update.Type,
+		})
+		if update.NewRoomEvent != nil {
+			eventType := update.NewRoomEvent.Event.Type()
+			logger = logger.WithFields(log.Fields{
+				"event_type":     eventType,
+				"event_id":       update.NewRoomEvent.Event.EventID(),
+				"adds_state":     len(update.NewRoomEvent.AddsStateEventIDs),
+				"removes_state":  len(update.NewRoomEvent.RemovesStateEventIDs),
+				"send_as_server": update.NewRoomEvent.SendAsServer,
+				"sender":         update.NewRoomEvent.Event.Sender(),
+			})
+			if update.NewRoomEvent.Event.StateKey() != nil {
+				logger = logger.WithField("state_key", *update.NewRoomEvent.Event.StateKey())
+			}
+			contentKey := keyContentFields[eventType]
+			if contentKey != "" {
+				value := gjson.GetBytes(update.NewRoomEvent.Event.Content(), contentKey)
+				if value.Exists() {
+					logger = logger.WithField("content_value", value.String())
+				}
+			}
+
+			if eventType == "m.room.server_acl" && update.NewRoomEvent.Event.StateKeyEquals("") {
+				ev := update.NewRoomEvent.Event.Unwrap()
+				defer r.ACLs.OnServerACLUpdate(ev)
+			}
+		}
+		logger.Tracef("Producing to topic '%s'", r.Topic)
+		if _, err := r.JetStream.PublishMsg(msg); err != nil {
+			logger.WithError(err).Errorf("Failed to produce to topic '%s': %s", r.Topic, err)
+			return err
+		}
+	}
+	return nil
+}

roomserver/roomserver.go

@@ -55,7 +55,6 @@ func NewInternalAPI(
 	return internal.NewRoomserverAPI(
 		base.ProcessContext, cfg, roomserverDB, js, nc,
 		cfg.Matrix.JetStream.Prefixed(jetstream.InputRoomEvent),
-		cfg.Matrix.JetStream.Prefixed(jetstream.OutputRoomEvent),
 		base.Caches, perspectiveServerNames,
 	)
 }

roomserver/storage/shared/storage.go

@@ -823,13 +823,39 @@ func (d *Database) handleRedactions(
 		return nil, "", nil
 	}
 
+	// Get the power level from the database, so we can verify the user is allowed to redact the event
+	powerLevels, err := d.GetStateEvent(ctx, event.RoomID(), gomatrixserverlib.MRoomPowerLevels, "")
+	if err != nil {
+		return nil, "", fmt.Errorf("d.GetStateEvent: %w", err)
+	}
+	pl, err := powerLevels.PowerLevels()
+	if err != nil {
+		return nil, "", fmt.Errorf("unable to get powerlevels for room: %w", err)
+	}
+
+	redactUser := pl.UserLevel(redactionEvent.Sender())
+	switch {
+	case redactUser >= pl.Redact:
+		// The power level of the redaction event’s sender is greater than or equal to the redact level.
+	case redactedEvent.Origin() == redactionEvent.Origin() && redactedEvent.Sender() == redactionEvent.Sender():
+		// The domain of the redaction event’s sender matches that of the original event’s sender.
+	default:
+		return nil, "", nil
+	}
+
 	// mark the event as redacted
+	if redactionsArePermanent {
+		redactedEvent.Event = redactedEvent.Redact()
+	}
+
 	err = redactedEvent.SetUnsignedField("redacted_because", redactionEvent)
 	if err != nil {
 		return nil, "", fmt.Errorf("redactedEvent.SetUnsignedField: %w", err)
 	}
-	if redactionsArePermanent {
-		redactedEvent.Event = redactedEvent.Redact()
+	// NOTSPEC: sytest relies on this unspecced field existing :(
+	err = redactedEvent.SetUnsignedField("redacted_by", redactionEvent.EventID())
+	if err != nil {
+		return nil, "", fmt.Errorf("redactedEvent.SetUnsignedField: %w", err)
 	}
 	// overwrite the eventJSON table
 	err = d.EventJSONTable.InsertEventJSON(ctx, txn, redactedEvent.EventNID, redactedEvent.JSON())

setup/base/sanity_unix.go

@@ -15,8 +15,21 @@ func platformSanityChecks() {
 	// If we run out of file descriptors, we might run into problems accessing
 	// PostgreSQL amongst other things. Complain at startup if we think the
 	// number of file descriptors is too low.
-	var rLimit syscall.Rlimit
-	if err := syscall.Getrlimit(syscall.RLIMIT_NOFILE, &rLimit); err == nil && rLimit.Cur < 65535 {
+	warn := func(rLimit *syscall.Rlimit) {
 		logrus.Warnf("IMPORTANT: Process file descriptor limit is currently %d, it is recommended to raise the limit for Dendrite to at least 65535 to avoid issues", rLimit.Cur)
 	}
+	var rLimit syscall.Rlimit
+	if err := syscall.Getrlimit(syscall.RLIMIT_NOFILE, &rLimit); err == nil && rLimit.Cur < 65535 {
+		// The file descriptor count is too low. Let's try to raise it.
+		rLimit.Cur = 65535
+		if err = syscall.Setrlimit(syscall.RLIMIT_NOFILE, &rLimit); err != nil {
+			// We failed to raise it, so log an error.
+			logrus.WithError(err).Warn("IMPORTANT: Failed to raise the file descriptor limit")
+			warn(&rLimit)
+		} else if err = syscall.Getrlimit(syscall.RLIMIT_NOFILE, &rLimit); err == nil && rLimit.Cur < 65535 {
+			// We think we successfully raised the limit, but a second call to
+			// get the limit told us that we didn't succeed. Log an error.
+			warn(&rLimit)
+		}
+	}
 }

setup/jetstream/streams.go

@@ -16,6 +16,7 @@ const (
 
 var (
 	InputRoomEvent          = "InputRoomEvent"
+	InputDeviceListUpdate   = "InputDeviceListUpdate"
 	OutputRoomEvent         = "OutputRoomEvent"
 	OutputSendToDeviceEvent = "OutputSendToDeviceEvent"
 	OutputKeyChangeEvent    = "OutputKeyChangeEvent"
@@ -45,6 +46,11 @@ var streams = []*nats.StreamConfig{
 		Retention: nats.InterestPolicy,
 		Storage:   nats.FileStorage,
 	},
+	{
+		Name:      InputDeviceListUpdate,
+		Retention: nats.InterestPolicy,
+		Storage:   nats.FileStorage,
+	},
 	{
 		Name:      OutputRoomEvent,
 		Retention: nats.InterestPolicy,

syncapi/consumers/presence.go

@@ -138,7 +138,7 @@ func (s *PresenceConsumer) onMessage(ctx context.Context, msg *nats.Msg) bool {
 	presence := msg.Header.Get("presence")
 	timestamp := msg.Header.Get("last_active_ts")
 	fromSync, _ := strconv.ParseBool(msg.Header.Get("from_sync"))
-	logrus.Debugf("syncAPI received presence event: %+v", msg.Header)
+	logrus.Tracef("syncAPI received presence event: %+v", msg.Header)
 
 	if fromSync { // do not process local presence changes; we already did this synchronously.
 		return true

syncapi/internal/keychange_test.go

@@ -43,9 +43,6 @@ func (k *mockKeyAPI) QueryOneTimeKeys(ctx context.Context, req *keyapi.QueryOneT
 }
 func (k *mockKeyAPI) QueryDeviceMessages(ctx context.Context, req *keyapi.QueryDeviceMessagesRequest, res *keyapi.QueryDeviceMessagesResponse) {
 
-}
-func (k *mockKeyAPI) InputDeviceListUpdate(ctx context.Context, req *keyapi.InputDeviceListUpdateRequest, res *keyapi.InputDeviceListUpdateResponse) {
-
 }
 func (k *mockKeyAPI) QuerySignatures(ctx context.Context, req *keyapi.QuerySignaturesRequest, res *keyapi.QuerySignaturesResponse) {
 }

syncapi/routing/context.go

@@ -15,6 +15,7 @@
 package routing
 
 import (
+	"context"
 	"database/sql"
 	"encoding/json"
 	"fmt"
@@ -25,6 +26,7 @@ import (
 	"github.com/matrix-org/dendrite/internal/caching"
 	roomserver "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"
 	"github.com/matrix-org/gomatrixserverlib"
 	"github.com/matrix-org/util"
@@ -97,7 +99,7 @@ func Context(
 	state, _ := syncDB.CurrentState(ctx, roomID, &stateFilter, nil)
 	// verify the user is allowed to see the context for this room/event
 	for _, x := range state {
-		var hisVis string
+		var hisVis gomatrixserverlib.HistoryVisibility
 		hisVis, err = x.HistoryVisibility()
 		if err != nil {
 			continue
@@ -149,13 +151,30 @@ func Context(
 	if len(response.State) > filter.Limit {
 		response.State = response.State[len(response.State)-filter.Limit:]
 	}
-
+	start, end, err := getStartEnd(ctx, syncDB, eventsBefore, eventsAfter)
+	if err == nil {
+		response.End = end.String()
+		response.Start = start.String()
+	}
 	return util.JSONResponse{
 		Code: http.StatusOK,
 		JSON: response,
 	}
 }
 
+func getStartEnd(ctx context.Context, syncDB storage.Database, startEvents, endEvents []*gomatrixserverlib.HeaderedEvent) (start, end types.TopologyToken, err error) {
+	if len(startEvents) > 0 {
+		start, err = syncDB.EventPositionInTopology(ctx, startEvents[0].EventID())
+		if err != nil {
+			return
+		}
+	}
+	if len(endEvents) > 0 {
+		end, err = syncDB.EventPositionInTopology(ctx, endEvents[0].EventID())
+	}
+	return
+}
+
 func applyLazyLoadMembers(
 	device *userapi.Device,
 	filter *gomatrixserverlib.RoomEventFilter,

syncapi/routing/messages.go

@@ -50,7 +50,7 @@ type messagesReq struct {
 type messagesResp struct {
 	Start       string                          `json:"start"`
 	StartStream string                          `json:"start_stream,omitempty"` // NOTSPEC: used by Cerulean, so clients can hit /messages then immediately /sync with a latest sync token
-	End         string                          `json:"end"`
+	End         string                          `json:"end,omitempty"`
 	Chunk       []gomatrixserverlib.ClientEvent `json:"chunk"`
 	State       []gomatrixserverlib.ClientEvent `json:"state"`
 }
@@ -200,30 +200,6 @@ func OnIncomingMessagesRequest(
 		return jsonerror.InternalServerError()
 	}
 
-	// at least fetch the membership events for the users returned in chunk if LazyLoadMembers is set
-	state := []gomatrixserverlib.ClientEvent{}
-	if filter.LazyLoadMembers {
-		membershipToUser := make(map[string]*gomatrixserverlib.HeaderedEvent)
-		for _, evt := range clientEvents {
-			// Don't add membership events the client should already know about
-			if _, cached := lazyLoadCache.IsLazyLoadedUserCached(device, roomID, evt.Sender); cached {
-				continue
-			}
-			membership, err := db.GetStateEvent(req.Context(), roomID, gomatrixserverlib.MRoomMember, evt.Sender)
-			if err != nil {
-				util.GetLogger(req.Context()).WithError(err).Error("failed to get membership event for user")
-				continue
-			}
-			if membership != nil {
-				membershipToUser[evt.Sender] = membership
-				lazyLoadCache.StoreLazyLoadedUser(device, roomID, evt.Sender, membership.EventID())
-			}
-		}
-		for _, evt := range membershipToUser {
-			state = append(state, gomatrixserverlib.HeaderedToClientEvent(evt, gomatrixserverlib.FormatSync))
-		}
-	}
-
 	util.GetLogger(req.Context()).WithFields(logrus.Fields{
 		"from":         from.String(),
 		"to":           to.String(),
@@ -237,7 +213,13 @@ func OnIncomingMessagesRequest(
 		Chunk: clientEvents,
 		Start: start.String(),
 		End:   end.String(),
-		State: state,
+	}
+	res.applyLazyLoadMembers(req.Context(), db, roomID, device, filter.LazyLoadMembers, lazyLoadCache)
+
+	// If we didn't return any events, set the end to an empty string, so it will be omitted
+	// in the response JSON.
+	if len(res.Chunk) == 0 {
+		res.End = ""
 	}
 	if fromStream != nil {
 		res.StartStream = fromStream.String()
@@ -250,6 +232,40 @@ func OnIncomingMessagesRequest(
 	}
 }
 
+// applyLazyLoadMembers loads membership events for users returned in Chunk, if the filter has
+// LazyLoadMembers enabled.
+func (m *messagesResp) applyLazyLoadMembers(
+	ctx context.Context,
+	db storage.Database,
+	roomID string,
+	device *userapi.Device,
+	lazyLoad bool,
+	lazyLoadCache caching.LazyLoadCache,
+) {
+	if !lazyLoad {
+		return
+	}
+	membershipToUser := make(map[string]*gomatrixserverlib.HeaderedEvent)
+	for _, evt := range m.Chunk {
+		// Don't add membership events the client should already know about
+		if _, cached := lazyLoadCache.IsLazyLoadedUserCached(device, roomID, evt.Sender); cached {
+			continue
+		}
+		membership, err := db.GetStateEvent(ctx, roomID, gomatrixserverlib.MRoomMember, evt.Sender)
+		if err != nil {
+			util.GetLogger(ctx).WithError(err).Error("failed to get membership event for user")
+			continue
+		}
+		if membership != nil {
+			membershipToUser[evt.Sender] = membership
+			lazyLoadCache.StoreLazyLoadedUser(device, roomID, evt.Sender, membership.EventID())
+		}
+	}
+	for _, evt := range membershipToUser {
+		m.State = append(m.State, gomatrixserverlib.HeaderedToClientEvent(evt, gomatrixserverlib.FormatSync))
+	}
+}
+
 func checkIsRoomForgotten(ctx context.Context, roomID, userID string, rsAPI api.SyncRoomserverAPI) (forgotten bool, exists bool, err error) {
 	req := api.QueryMembershipForUserRequest{
 		RoomID: roomID,

syncapi/streams/stream_presence.go

@@ -111,7 +111,7 @@ func (p *PresenceStreamProvider) IncrementalSync(
 			currentlyActive := prevPresence.CurrentlyActive()
 			skip := prevPresence.Equals(presence) && currentlyActive && req.Device.UserID != presence.UserID
 			if skip {
-				req.Log.Debugf("Skipping presence, no change (%s)", presence.UserID)
+				req.Log.Tracef("Skipping presence, no change (%s)", presence.UserID)
 				continue
 			}
 		}

sytest-blacklist

@@ -48,3 +48,4 @@ Notifications can be viewed with GET /notifications
 # More flakey
 
 If remote user leaves room we no longer receive device updates
+Guest users can join guest_access rooms

sytest-whitelist

@@ -241,7 +241,6 @@ Inbound federation can receive v2 /send_join
 Message history can be paginated
 Backfill works correctly with history visibility set to joined
 Guest user cannot call /events globally
-Guest users can join guest_access rooms
 Guest user can set display names
 Guest user cannot upgrade other users
 Guest non-joined user cannot call /events on shared room
@@ -720,3 +719,4 @@ registration is idempotent, with username specified
 Setting state twice is idempotent
 Joining room twice is idempotent
 Inbound federation can return missing events for shared visibility
+Inbound federation ignores redactions from invalid servers room > v3

userapi/internal/api.go

@@ -33,6 +33,7 @@ import (
 	"github.com/matrix-org/dendrite/internal/pushrules"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
 	keyapi "github.com/matrix-org/dendrite/keyserver/api"
+	rsapi "github.com/matrix-org/dendrite/roomserver/api"
 	"github.com/matrix-org/dendrite/setup/config"
 	"github.com/matrix-org/dendrite/userapi/api"
 	"github.com/matrix-org/dendrite/userapi/producers"
@@ -49,6 +50,7 @@ type UserInternalAPI struct {
 	// AppServices is the list of all registered AS
 	AppServices []config.ApplicationService
 	KeyAPI      keyapi.UserKeyAPI
+	RSAPI       rsapi.UserRoomserverAPI
 }
 
 func (a *UserInternalAPI) InputAccountData(ctx context.Context, req *api.InputAccountDataRequest, res *api.InputAccountDataResponse) error {
@@ -452,6 +454,30 @@ func (a *UserInternalAPI) queryAppServiceToken(ctx context.Context, token, appSe
 
 // PerformAccountDeactivation deactivates the user's account, removing all ability for the user to login again.
 func (a *UserInternalAPI) PerformAccountDeactivation(ctx context.Context, req *api.PerformAccountDeactivationRequest, res *api.PerformAccountDeactivationResponse) error {
+	evacuateReq := &rsapi.PerformAdminEvacuateUserRequest{
+		UserID: fmt.Sprintf("@%s:%s", req.Localpart, a.ServerName),
+	}
+	evacuateRes := &rsapi.PerformAdminEvacuateUserResponse{}
+	a.RSAPI.PerformAdminEvacuateUser(ctx, evacuateReq, evacuateRes)
+	if err := evacuateRes.Error; err != nil {
+		logrus.WithError(err).Errorf("Failed to evacuate user after account deactivation")
+	}
+
+	deviceReq := &api.PerformDeviceDeletionRequest{
+		UserID: fmt.Sprintf("@%s:%s", req.Localpart, a.ServerName),
+	}
+	deviceRes := &api.PerformDeviceDeletionResponse{}
+	if err := a.PerformDeviceDeletion(ctx, deviceReq, deviceRes); err != nil {
+		return err
+	}
+
+	pusherReq := &api.PerformPusherDeletionRequest{
+		Localpart: req.Localpart,
+	}
+	if err := a.PerformPusherDeletion(ctx, pusherReq, &struct{}{}); err != nil {
+		return err
+	}
+
 	err := a.DB.DeactivateAccount(ctx, req.Localpart)
 	res.AccountDeactivated = err == nil
 	return err

userapi/userapi.go

@@ -78,6 +78,7 @@ func NewInternalAPI(
 		ServerName:           cfg.Matrix.ServerName,
 		AppServices:          appServices,
 		KeyAPI:               keyAPI,
+		RSAPI:                rsAPI,
 		DisableTLSValidation: cfg.PushGatewayDisableTLSValidation,
 	}