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

README.md

@@ -21,8 +21,7 @@ As of October 2020 (current [progress below](#progress)), Dendrite has now enter
 This does not mean:
 
 - Dendrite is bug-free. It has not yet been battle-tested in the real world and so will be error prone initially.
-- All of the CS/Federation APIs are implemented. We are tracking progress via a script called 'Are We Synapse Yet?'. In particular,
-  presence and push notifications are entirely missing from Dendrite. See [CHANGES.md](CHANGES.md) for updates.
+- Dendrite is feature-complete. There may be client or federation APIs that are not implemented.
 - Dendrite is ready for massive homeserver deployments. You cannot shard each microservice, only run each one on a different machine.
 
 Currently, we expect Dendrite to function well for small (10s/100s of users) homeserver deployments as well as P2P Matrix nodes in-browser or on mobile devices.
@@ -36,6 +35,9 @@ If you have further questions, please take a look at [our FAQ](docs/FAQ.md) or j
 
 ## Requirements
 
+See the [Planning your Installation](https://matrix-org.github.io/dendrite/installation/planning) page for
+more information on requirements.
+
 To build Dendrite, you will need Go 1.18 or later.
 
 For a usable federating Dendrite deployment, you will also need:
@@ -83,11 +85,11 @@ $ ./bin/create-account --config dendrite.yaml -username alice
 
 Then point your favourite Matrix client at `http://localhost:8008` or `https://localhost:8448`.
 
-## <a id="progress"></a> Progress
+## Progress
 
 We use a script called Are We Synapse Yet which checks Sytest compliance rates. Sytest is a black-box homeserver
 test rig with around 900 tests. The script works out how many of these tests are passing on Dendrite and it
-updates with CI. As of April 2022 we're at around 83% CS API coverage and 95% Federation coverage, though check
+updates with CI. As of August 2022 we're at around 83% CS API coverage and 95% Federation coverage, though check
 CI for the latest numbers. In practice, this means you can communicate locally and via federation with Synapse
 servers such as matrix.org reasonably well, although there are still some missing features (like Search).
 
@@ -119,53 +121,8 @@ We would be grateful for any help on issues marked as
 all have related Sytests which need to pass in order for the issue to be closed. Once you've written your
 code, you can quickly run Sytest to ensure that the test names are now passing.
 
-For example, if the test `Local device key changes get to remote servers` was marked as failing, find the
-test file (e.g via `grep` or via the
-[CI log output](https://buildkite.com/matrix-dot-org/dendrite/builds/2826#39cff5de-e032-4ad0-ad26-f819e6919c42)
-it's `tests/50federation/40devicelists.pl` ) then to run Sytest:
-
-```
-docker run --rm --name sytest
--v "/Users/kegan/github/sytest:/sytest"
--v "/Users/kegan/github/dendrite:/src"
--v "/Users/kegan/logs:/logs"
--v "/Users/kegan/go/:/gopath"
--e "POSTGRES=1" -e "DENDRITE_TRACE_HTTP=1"
-matrixdotorg/sytest-dendrite:latest tests/50federation/40devicelists.pl
-```
-
-See [sytest.md](docs/sytest.md) for the full description of these flags.
-
-You can try running sytest outside of docker for faster runs, but the dependencies can be temperamental
-and we recommend using docker where possible.
-
-```
-cd sytest
-export PERL5LIB=$HOME/lib/perl5
-export PERL_MB_OPT=--install_base=$HOME
-export PERL_MM_OPT=INSTALL_BASE=$HOME
-./install-deps.pl
-
-./run-tests.pl -I Dendrite::Monolith -d $PATH_TO_DENDRITE_BINARIES
-```
-
-Sometimes Sytest is testing the wrong thing or is flakey, so it will need to be patched.
-Ask on `#dendrite-dev:matrix.org` if you think this is the case for you and we'll be happy to help.
-
-If you're new to the project, see [CONTRIBUTING.md](docs/CONTRIBUTING.md) to get up to speed then
+If you're new to the project, see our
+[Contributing page](https://matrix-org.github.io/dendrite/development/contributing) to get up to speed, then
 look for [Good First Issues](https://github.com/matrix-org/dendrite/labels/good%20first%20issue). If you're
 familiar with the project, look for [Help Wanted](https://github.com/matrix-org/dendrite/labels/help-wanted)
 issues.
-
-## Hardware requirements
-
-Dendrite in Monolith + SQLite works in a range of environments including iOS and in-browser via WASM.
-
-For small homeserver installations joined on ~10s rooms on matrix.org with ~100s of users in those rooms, including some
-encrypted rooms:
-
-- Memory: uses around 100MB of RAM, with peaks at around 200MB.
-- Disk space: After a few months of usage, the database grew to around 2GB (in Monolith mode).
-- CPU: Brief spikes when processing events, typically idles at 1% CPU.
-
-This means Dendrite should comfortably work on things like Raspberry Pis.

clientapi/auth/login.go

@@ -18,7 +18,6 @@ import (
 	"context"
 	"encoding/json"
 	"io"
-	"io/ioutil"
 	"net/http"
 
 	"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
@@ -34,7 +33,7 @@ import (
 // If the final return value is non-nil, an error occurred and the cleanup function
 // is nil.
 func LoginFromJSONReader(ctx context.Context, r io.Reader, useraccountAPI uapi.UserLoginAPI, userAPI UserInternalAPIForLogin, cfg *config.ClientAPI) (*Login, LoginCleanupFunc, *util.JSONResponse) {
-	reqBytes, err := ioutil.ReadAll(r)
+	reqBytes, err := io.ReadAll(r)
 	if err != nil {
 		err := &util.JSONResponse{
 			Code: http.StatusBadRequest,

clientapi/httputil/httputil.go

@@ -16,7 +16,7 @@ package httputil
 
 import (
 	"encoding/json"
-	"io/ioutil"
+	"io"
 	"net/http"
 	"unicode/utf8"
 
@@ -29,9 +29,9 @@ import (
 func UnmarshalJSONRequest(req *http.Request, iface interface{}) *util.JSONResponse {
 	// encoding/json allows invalid utf-8, matrix does not
 	// https://matrix.org/docs/spec/client_server/r0.6.1#api-standards
-	body, err := ioutil.ReadAll(req.Body)
+	body, err := io.ReadAll(req.Body)
 	if err != nil {
-		util.GetLogger(req.Context()).WithError(err).Error("ioutil.ReadAll failed")
+		util.GetLogger(req.Context()).WithError(err).Error("io.ReadAll failed")
 		resp := jsonerror.InternalServerError()
 		return &resp
 	}

clientapi/routing/account_data.go

@@ -17,7 +17,7 @@ package routing
 import (
 	"encoding/json"
 	"fmt"
-	"io/ioutil"
+	"io"
 	"net/http"
 
 	"github.com/matrix-org/dendrite/clientapi/httputil"
@@ -101,9 +101,9 @@ func SaveAccountData(
 		}
 	}
 
-	body, err := ioutil.ReadAll(req.Body)
+	body, err := io.ReadAll(req.Body)
 	if err != nil {
-		util.GetLogger(req.Context()).WithError(err).Error("ioutil.ReadAll failed")
+		util.GetLogger(req.Context()).WithError(err).Error("io.ReadAll failed")
 		return jsonerror.InternalServerError()
 	}
 

clientapi/routing/deactivate.go

@@ -1,7 +1,7 @@
 package routing
 
 import (
-	"io/ioutil"
+	"io"
 	"net/http"
 
 	"github.com/matrix-org/dendrite/clientapi/auth"
@@ -20,7 +20,7 @@ func Deactivate(
 ) util.JSONResponse {
 	ctx := req.Context()
 	defer req.Body.Close() // nolint:errcheck
-	bodyBytes, err := ioutil.ReadAll(req.Body)
+	bodyBytes, err := io.ReadAll(req.Body)
 	if err != nil {
 		return util.JSONResponse{
 			Code: http.StatusBadRequest,

clientapi/routing/device.go

@@ -15,7 +15,7 @@
 package routing
 
 import (
-	"io/ioutil"
+	"io"
 	"net"
 	"net/http"
 
@@ -175,7 +175,7 @@ func DeleteDeviceById(
 	}()
 	ctx := req.Context()
 	defer req.Body.Close() // nolint:errcheck
-	bodyBytes, err := ioutil.ReadAll(req.Body)
+	bodyBytes, err := io.ReadAll(req.Body)
 	if err != nil {
 		return util.JSONResponse{
 			Code: http.StatusBadRequest,

clientapi/routing/register.go

@@ -19,7 +19,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"io/ioutil"
+	"io"
 	"net/http"
 	"net/url"
 	"regexp"
@@ -371,7 +371,7 @@ func validateRecaptcha(
 
 	// Grab the body of the response from the captcha server
 	var r recaptchaResponse
-	body, err := ioutil.ReadAll(resp.Body)
+	body, err := io.ReadAll(resp.Body)
 	if err != nil {
 		return &util.JSONResponse{
 			Code: http.StatusGatewayTimeout,
@@ -539,7 +539,7 @@ func Register(
 	cfg *config.ClientAPI,
 ) util.JSONResponse {
 	defer req.Body.Close() // nolint: errcheck
-	reqBody, err := ioutil.ReadAll(req.Body)
+	reqBody, err := io.ReadAll(req.Body)
 	if err != nil {
 		return util.JSONResponse{
 			Code: http.StatusBadRequest,

clientapi/routing/register_secret_test.go

@@ -2,7 +2,7 @@ package routing
 
 import (
 	"bytes"
-	"io/ioutil"
+	"io"
 	"testing"
 
 	"github.com/patrickmn/go-cache"
@@ -13,7 +13,7 @@ func TestSharedSecretRegister(t *testing.T) {
 	jsonStr := []byte(`{"admin":false,"mac":"f1ba8d37123866fd659b40de4bad9b0f8965c565","nonce":"759f047f312b99ff428b21d581256f8592b8976e58bc1b543972dc6147e529a79657605b52d7becd160ff5137f3de11975684319187e06901955f79e5a6c5a79","password":"wonderland","username":"alice"}`)
 	sharedSecret := "dendritetest"
 
-	req, err := NewSharedSecretRegistrationRequest(ioutil.NopCloser(bytes.NewBuffer(jsonStr)))
+	req, err := NewSharedSecretRegistrationRequest(io.NopCloser(bytes.NewBuffer(jsonStr)))
 	if err != nil {
 		t.Fatalf("failed to read request: %s", err)
 	}

cmd/create-account/main.go

@@ -19,7 +19,6 @@ import (
 	"flag"
 	"fmt"
 	"io"
-	"io/ioutil"
 	"os"
 	"regexp"
 	"strings"
@@ -157,7 +156,7 @@ func main() {
 func getPassword(password, pwdFile string, pwdStdin bool, r io.Reader) (string, error) {
 	// read password from file
 	if pwdFile != "" {
-		pw, err := ioutil.ReadFile(pwdFile)
+		pw, err := os.ReadFile(pwdFile)
 		if err != nil {
 			return "", fmt.Errorf("Unable to read password from file: %v", err)
 		}
@@ -166,7 +165,7 @@ func getPassword(password, pwdFile string, pwdStdin bool, r io.Reader) (string,
 
 	// read password from stdin
 	if pwdStdin {
-		data, err := ioutil.ReadAll(r)
+		data, err := io.ReadAll(r)
 		if err != nil {
 			return "", fmt.Errorf("Unable to read password from stdin: %v", err)
 		}

cmd/dendrite-demo-pinecone/main.go

@@ -21,7 +21,6 @@ import (
 	"encoding/hex"
 	"flag"
 	"fmt"
-	"io/ioutil"
 	"net"
 	"net/http"
 	"os"
@@ -76,11 +75,11 @@ func main() {
 		if pk, sk, err = ed25519.GenerateKey(nil); err != nil {
 			panic(err)
 		}
-		if err = ioutil.WriteFile(keyfile, sk, 0644); err != nil {
+		if err = os.WriteFile(keyfile, sk, 0644); err != nil {
 			panic(err)
 		}
 	} else if err == nil {
-		if sk, err = ioutil.ReadFile(keyfile); err != nil {
+		if sk, err = os.ReadFile(keyfile); err != nil {
 			panic(err)
 		}
 		if len(sk) != ed25519.PrivateKeySize {

cmd/dendrite-demo-yggdrasil/yggconn/node.go

@@ -20,7 +20,6 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
-	"io/ioutil"
 	"log"
 	"net"
 	"os"
@@ -69,7 +68,7 @@ func Setup(instanceName, storageDirectory, peerURI string) (*Node, error) {
 
 	yggfile := fmt.Sprintf("%s/%s-yggdrasil.conf", storageDirectory, instanceName)
 	if _, err := os.Stat(yggfile); !os.IsNotExist(err) {
-		yggconf, e := ioutil.ReadFile(yggfile)
+		yggconf, e := os.ReadFile(yggfile)
 		if e != nil {
 			panic(err)
 		}
@@ -88,7 +87,7 @@ func Setup(instanceName, storageDirectory, peerURI string) (*Node, error) {
 	if err != nil {
 		panic(err)
 	}
-	if e := ioutil.WriteFile(yggfile, j, 0600); e != nil {
+	if e := os.WriteFile(yggfile, j, 0600); e != nil {
 		n.log.Printf("Couldn't write private key to file '%s': %s\n", yggfile, e)
 	}
 

cmd/dendrite-upgrade-tests/main.go

@@ -6,7 +6,7 @@ import (
 	"encoding/json"
 	"flag"
 	"fmt"
-	"io/ioutil"
+	"io"
 	"log"
 	"net/http"
 	"os"
@@ -128,7 +128,7 @@ func downloadArchive(cli *http.Client, tmpDir, archiveURL string, dockerfile []b
 		return nil, err
 	}
 	// add top level Dockerfile
-	err = ioutil.WriteFile(path.Join(tmpDir, "Dockerfile"), dockerfile, os.ModePerm)
+	err = os.WriteFile(path.Join(tmpDir, "Dockerfile"), dockerfile, os.ModePerm)
 	if err != nil {
 		return nil, fmt.Errorf("failed to inject /Dockerfile: %w", err)
 	}
@@ -150,7 +150,7 @@ func buildDendrite(httpClient *http.Client, dockerClient *client.Client, tmpDir,
 	if branchOrTagName == HEAD && *flagHead != "" {
 		log.Printf("%s: Using %s as HEAD", branchOrTagName, *flagHead)
 		// add top level Dockerfile
-		err = ioutil.WriteFile(path.Join(*flagHead, "Dockerfile"), []byte(Dockerfile), os.ModePerm)
+		err = os.WriteFile(path.Join(*flagHead, "Dockerfile"), []byte(Dockerfile), os.ModePerm)
 		if err != nil {
 			return "", fmt.Errorf("custom HEAD: failed to inject /Dockerfile: %w", err)
 		}
@@ -388,7 +388,7 @@ func runImage(dockerClient *client.Client, volumeName, version, imageID string)
 		})
 		// ignore errors when cannot get logs, it's just for debugging anyways
 		if err == nil {
-			logbody, err := ioutil.ReadAll(logs)
+			logbody, err := io.ReadAll(logs)
 			if err == nil {
 				log.Printf("Container logs:\n\n%s\n\n", string(logbody))
 			}

cmd/furl/main.go

@@ -9,7 +9,6 @@ import (
 	"encoding/pem"
 	"flag"
 	"fmt"
-	"io/ioutil"
 	"net/url"
 	"os"
 
@@ -30,7 +29,7 @@ func main() {
 		os.Exit(1)
 	}
 
-	data, err := ioutil.ReadFile(*requestKey)
+	data, err := os.ReadFile(*requestKey)
 	if err != nil {
 		panic(err)
 	}

dendrite-sample.monolith.yaml

@@ -178,13 +178,16 @@ client_api:
 
   # TURN server information that this homeserver should send to clients.
   turn:
-    turn_user_lifetime: ""
+    turn_user_lifetime: "5m"
     turn_uris:
     #  - turn:turn.server.org?transport=udp
     #  - turn:turn.server.org?transport=tcp
     turn_shared_secret: ""
-    turn_username: ""
-    turn_password: ""
+    # If your TURN server requires static credentials, then you will need to enter
+    # them here instead of supplying a shared secret. Note that these credentials
+    # will be visible to clients!
+    # turn_username: ""
+    # turn_password: ""
 
   # Settings for rate-limited endpoints. Rate limiting kicks in after the threshold
   # number of "slots" have been taken by requests from a specific host. Each "slot"

dendrite-sample.polylith.yaml

@@ -181,13 +181,16 @@ client_api:
 
   # TURN server information that this homeserver should send to clients.
   turn:
-    turn_user_lifetime: ""
+    turn_user_lifetime: "5m"
     turn_uris:
     #  - turn:turn.server.org?transport=udp
     #  - turn:turn.server.org?transport=tcp
     turn_shared_secret: ""
-    turn_username: ""
-    turn_password: ""
+    # If your TURN server requires static credentials, then you will need to enter
+    # them here instead of supplying a shared secret. Note that these credentials
+    # will be visible to clients!
+    # turn_username: ""
+    # turn_password: ""
 
   # Settings for rate-limited endpoints. Rate limiting kicks in after the threshold
   # number of "slots" have been taken by requests from a specific host. Each "slot"

federationapi/federationapi_keys_test.go

@@ -6,7 +6,7 @@ import (
 	"crypto/ed25519"
 	"encoding/json"
 	"fmt"
-	"io/ioutil"
+	"io"
 	"net/http"
 	"os"
 	"testing"
@@ -66,7 +66,7 @@ func TestMain(m *testing.M) {
 			s.cache = caching.NewRistrettoCache(8*1024*1024, time.Hour, false)
 
 			// Create a temporary directory for JetStream.
-			d, err := ioutil.TempDir("./", "jetstream*")
+			d, err := os.MkdirTemp("./", "jetstream*")
 			if err != nil {
 				panic(err)
 			}
@@ -136,7 +136,7 @@ func (m *MockRoundTripper) RoundTrip(req *http.Request) (res *http.Response, err
 	// And respond.
 	res = &http.Response{
 		StatusCode: 200,
-		Body:       ioutil.NopCloser(bytes.NewReader(body)),
+		Body:       io.NopCloser(bytes.NewReader(body)),
 	}
 	return
 }

internal/caching/cache_lazy_load_members.go

@@ -14,6 +14,7 @@ type lazyLoadingCacheKey struct {
 type LazyLoadCache interface {
 	StoreLazyLoadedUser(device *userapi.Device, roomID, userID, eventID string)
 	IsLazyLoadedUserCached(device *userapi.Device, roomID, userID string) (string, bool)
+	InvalidateLazyLoadedUser(device *userapi.Device, roomID, userID string)
 }
 
 func (c Caches) StoreLazyLoadedUser(device *userapi.Device, roomID, userID, eventID string) {
@@ -33,3 +34,12 @@ func (c Caches) IsLazyLoadedUserCached(device *userapi.Device, roomID, userID st
 		TargetUserID: userID,
 	})
 }
+
+func (c Caches) InvalidateLazyLoadedUser(device *userapi.Device, roomID, userID string) {
+	c.LazyLoading.Unset(lazyLoadingCacheKey{
+		UserID:       device.UserID,
+		DeviceID:     device.ID,
+		RoomID:       roomID,
+		TargetUserID: userID,
+	})
+}

internal/log_unix.go

@@ -18,7 +18,7 @@
 package internal
 
 import (
-	"io/ioutil"
+	"io"
 	"log/syslog"
 
 	"github.com/MFAshby/stdemuxerhook"
@@ -63,7 +63,7 @@ func SetupHookLogging(hooks []config.LogrusHook, componentName string) {
 		setupStdLogHook(logrus.InfoLevel)
 	}
 	// Hooks are now configured for stdout/err, so throw away the default logger output
-	logrus.SetOutput(ioutil.Discard)
+	logrus.SetOutput(io.Discard)
 }
 
 func checkSyslogHookParams(params map[string]interface{}) {

keyserver/internal/device_list_update_test.go

@@ -18,7 +18,7 @@ import (
 	"context"
 	"crypto/ed25519"
 	"fmt"
-	"io/ioutil"
+	"io"
 	"net/http"
 	"net/url"
 	"reflect"
@@ -203,7 +203,7 @@ func TestUpdateNoPrevID(t *testing.T) {
 		}
 		return &http.Response{
 			StatusCode: 200,
-			Body: ioutil.NopCloser(strings.NewReader(`
+			Body: io.NopCloser(strings.NewReader(`
 			{
 				"user_id": "` + remoteUserID + `",
 				"stream_id": 5,
@@ -318,7 +318,7 @@ func TestDebounce(t *testing.T) {
 	// now send the response over federation
 	fedCh <- &http.Response{
 		StatusCode: 200,
-		Body: ioutil.NopCloser(strings.NewReader(`
+		Body: io.NopCloser(strings.NewReader(`
 		{
 			"user_id": "` + userID + `",
 			"stream_id": 5,

keyserver/storage/postgres/key_changes_table.go

@@ -18,6 +18,8 @@ import (
 	"context"
 	"database/sql"
 
+	"github.com/lib/pq"
+
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/keyserver/storage/postgres/deltas"
@@ -64,7 +66,8 @@ func NewPostgresKeyChangesTable(db *sql.DB) (tables.KeyChanges, error) {
 	// TODO: Remove when we are sure we are not having goose artefacts in the db
 	// This forces an error, which indicates the migration is already applied, since the
 	// column partition was removed from the table
-	err = db.QueryRow("SELECT partition FROM keyserver_key_changes LIMIT 1;").Scan()
+	var count int
+	err = db.QueryRow("SELECT partition FROM keyserver_key_changes LIMIT 1;").Scan(&count)
 	if err == nil {
 		m := sqlutil.NewMigrator(db)
 		m.AddMigrations(sqlutil.Migration{
@@ -72,6 +75,16 @@ func NewPostgresKeyChangesTable(db *sql.DB) (tables.KeyChanges, error) {
 			Up:      deltas.UpRefactorKeyChanges,
 		})
 		return s, m.Up(context.Background())
+	} else {
+		switch e := err.(type) {
+		case *pq.Error:
+			// ignore undefined_column (42703) errors, as this is expected at this point
+			if e.Code != "42703" {
+				return nil, err
+			}
+		default:
+			return nil, err
+		}
 	}
 	return s, nil
 }

keyserver/storage/sqlite3/key_changes_table.go

@@ -61,7 +61,8 @@ func NewSqliteKeyChangesTable(db *sql.DB) (tables.KeyChanges, error) {
 	// TODO: Remove when we are sure we are not having goose artefacts in the db
 	// This forces an error, which indicates the migration is already applied, since the
 	// column partition was removed from the table
-	err = db.QueryRow("SELECT partition FROM keyserver_key_changes LIMIT 1;").Scan()
+	var count int
+	err = db.QueryRow("SELECT partition FROM keyserver_key_changes LIMIT 1;").Scan(&count)
 	if err == nil {
 		m := sqlutil.NewMigrator(db)
 		m.AddMigrations(sqlutil.Migration{

mediaapi/fileutils/fileutils.go

@@ -21,7 +21,6 @@ import (
 	"encoding/base64"
 	"fmt"
 	"io"
-	"io/ioutil"
 	"os"
 	"path/filepath"
 	"strings"
@@ -180,7 +179,7 @@ func createTempDir(baseDirectory config.Path) (types.Path, error) {
 	if err := os.MkdirAll(baseTmpDir, 0770); err != nil {
 		return "", fmt.Errorf("failed to create base temp dir: %w", err)
 	}
-	tmpDir, err := ioutil.TempDir(baseTmpDir, "")
+	tmpDir, err := os.MkdirTemp(baseTmpDir, "")
 	if err != nil {
 		return "", fmt.Errorf("failed to create temp dir: %w", err)
 	}

mediaapi/routing/download.go

@@ -19,7 +19,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
-	"io/ioutil"
 	"mime"
 	"net/http"
 	"net/url"
@@ -695,7 +694,7 @@ func (r *downloadRequest) GetContentLengthAndReader(contentLengthHeader string,
 
 		// We successfully parsed the Content-Length, so we'll return a limited
 		// reader that restricts us to reading only up to this size.
-		reader = ioutil.NopCloser(io.LimitReader(*body, parsedLength))
+		reader = io.NopCloser(io.LimitReader(*body, parsedLength))
 		contentLength = parsedLength
 	} else {
 		// Content-Length header is missing. If we have a maximum file size
@@ -704,7 +703,7 @@ func (r *downloadRequest) GetContentLengthAndReader(contentLengthHeader string,
 		// ultimately it will get rewritten later when the temp file is written
 		// to disk.
 		if maxFileSizeBytes > 0 {
-			reader = ioutil.NopCloser(io.LimitReader(*body, int64(maxFileSizeBytes)))
+			reader = io.NopCloser(io.LimitReader(*body, int64(maxFileSizeBytes)))
 		}
 		contentLength = 0
 	}

roomserver/storage/postgres/storage.go

@@ -19,8 +19,10 @@ import (
 	"database/sql"
 	"fmt"
 
+	"github.com/lib/pq"
 	// Import the postgres database driver.
 	_ "github.com/lib/pq"
+
 	"github.com/matrix-org/dendrite/internal/caching"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/roomserver/storage/postgres/deltas"
@@ -53,21 +55,32 @@ func Open(base *base.BaseDendrite, dbProperties *config.DatabaseOptions, cache c
 	// TODO: Remove when we are sure we are not having goose artefacts in the db
 	// This forces an error, which indicates the migration is already applied, since the
 	// column event_nid was removed from the table
-	err = db.QueryRow("SELECT event_nid FROM roomserver_state_block LIMIT 1;").Scan()
+	var eventNID int
+	err = db.QueryRow("SELECT event_nid FROM roomserver_state_block LIMIT 1;").Scan(&eventNID)
 	if err == nil {
 		m := sqlutil.NewMigrator(db)
 		m.AddMigrations(sqlutil.Migration{
 			Version: "roomserver: state blocks refactor",
 			Up:      deltas.UpStateBlocksRefactor,
 		})
-		if err := m.Up(base.Context()); err != nil {
+		if err = m.Up(base.Context()); err != nil {
+			return nil, err
+		}
+	} else {
+		switch e := err.(type) {
+		case *pq.Error:
+			// ignore undefined_column (42703) errors, as this is expected at this point
+			if e.Code != "42703" {
+				return nil, err
+			}
+		default:
 			return nil, err
 		}
 	}
 
 	// Then prepare the statements. Now that the migrations have run, any columns referred
 	// to in the database code should now exist.
-	if err := d.prepare(db, writer, cache); err != nil {
+	if err = d.prepare(db, writer, cache); err != nil {
 		return nil, err
 	}
 

roomserver/storage/sqlite3/storage.go

@@ -20,6 +20,8 @@ import (
 	"database/sql"
 	"fmt"
 
+	"github.com/matrix-org/gomatrixserverlib"
+
 	"github.com/matrix-org/dendrite/internal/caching"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/roomserver/storage/shared"
@@ -27,7 +29,6 @@ import (
 	"github.com/matrix-org/dendrite/roomserver/types"
 	"github.com/matrix-org/dendrite/setup/base"
 	"github.com/matrix-org/dendrite/setup/config"
-	"github.com/matrix-org/gomatrixserverlib"
 )
 
 // A Database is used to store room events and stream offsets.
@@ -63,21 +64,22 @@ func Open(base *base.BaseDendrite, dbProperties *config.DatabaseOptions, cache c
 	// TODO: Remove when we are sure we are not having goose artefacts in the db
 	// This forces an error, which indicates the migration is already applied, since the
 	// column event_nid was removed from the table
-	err = db.QueryRow("SELECT event_nid FROM roomserver_state_block LIMIT 1;").Scan()
+	var eventNID int
+	err = db.QueryRow("SELECT event_nid FROM roomserver_state_block LIMIT 1;").Scan(&eventNID)
 	if err == nil {
 		m := sqlutil.NewMigrator(db)
 		m.AddMigrations(sqlutil.Migration{
 			Version: "roomserver: state blocks refactor",
 			Up:      deltas.UpStateBlocksRefactor,
 		})
-		if err := m.Up(base.Context()); err != nil {
+		if err = m.Up(base.Context()); err != nil {
 			return nil, err
 		}
 	}
 
 	// Then prepare the statements. Now that the migrations have run, any columns referred
 	// to in the database code should now exist.
-	if err := d.prepare(db, writer, cache); err != nil {
+	if err = d.prepare(db, writer, cache); err != nil {
 		return nil, err
 	}
 

setup/config/config.go

@@ -19,8 +19,8 @@ import (
 	"encoding/pem"
 	"fmt"
 	"io"
-	"io/ioutil"
 	"net/url"
+	"os"
 	"path/filepath"
 	"regexp"
 	"strings"
@@ -191,7 +191,7 @@ type ConfigErrors []string
 // Load a yaml config file for a server run as multiple processes or as a monolith.
 // Checks the config to ensure that it is valid.
 func Load(configPath string, monolith bool) (*Dendrite, error) {
-	configData, err := ioutil.ReadFile(configPath)
+	configData, err := os.ReadFile(configPath)
 	if err != nil {
 		return nil, err
 	}
@@ -199,9 +199,9 @@ func Load(configPath string, monolith bool) (*Dendrite, error) {
 	if err != nil {
 		return nil, err
 	}
-	// Pass the current working directory and ioutil.ReadFile so that they can
+	// Pass the current working directory and os.ReadFile so that they can
 	// be mocked in the tests
-	return loadConfig(basePath, configData, ioutil.ReadFile, monolith)
+	return loadConfig(basePath, configData, os.ReadFile, monolith)
 }
 
 func loadConfig(
@@ -530,7 +530,7 @@ func (config *Dendrite) KeyServerURL() string {
 // SetupTracing configures the opentracing using the supplied configuration.
 func (config *Dendrite) SetupTracing(serviceName string) (closer io.Closer, err error) {
 	if !config.Tracing.Enabled {
-		return ioutil.NopCloser(bytes.NewReader([]byte{})), nil
+		return io.NopCloser(bytes.NewReader([]byte{})), nil
 	}
 	return config.Tracing.Jaeger.InitGlobalTracer(
 		serviceName,

setup/config/config_appservice.go

@@ -16,7 +16,7 @@ package config
 
 import (
 	"fmt"
-	"io/ioutil"
+	"os"
 	"path/filepath"
 	"regexp"
 	"strings"
@@ -181,7 +181,7 @@ func loadAppServices(config *AppServiceAPI, derived *Derived) error {
 		}
 
 		// Read the application service's config file
-		configData, err := ioutil.ReadFile(absPath)
+		configData, err := os.ReadFile(absPath)
 		if err != nil {
 			return err
 		}

setup/mscs/msc2836/msc2836_test.go

@@ -7,7 +7,7 @@ import (
 	"crypto/sha256"
 	"encoding/base64"
 	"encoding/json"
-	"io/ioutil"
+	"io"
 	"net/http"
 	"sort"
 	"strings"
@@ -428,12 +428,12 @@ func postRelationships(t *testing.T, expectCode int, accessToken string, req *ms
 		t.Fatalf("failed to do request: %s", err)
 	}
 	if res.StatusCode != expectCode {
-		body, _ := ioutil.ReadAll(res.Body)
+		body, _ := io.ReadAll(res.Body)
 		t.Fatalf("wrong response code, got %d want %d - body: %s", res.StatusCode, expectCode, string(body))
 	}
 	if res.StatusCode == 200 {
 		var result msc2836.EventRelationshipResponse
-		body, err := ioutil.ReadAll(res.Body)
+		body, err := io.ReadAll(res.Body)
 		if err != nil {
 			t.Fatalf("response 200 OK but failed to read response body: %s", err)
 		}

syncapi/consumers/clientapi.go

@@ -21,6 +21,11 @@ import (
 	"fmt"
 
 	"github.com/getsentry/sentry-go"
+	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/nats-io/nats.go"
+	"github.com/sirupsen/logrus"
+	log "github.com/sirupsen/logrus"
+
 	"github.com/matrix-org/dendrite/internal/eventutil"
 	"github.com/matrix-org/dendrite/setup/config"
 	"github.com/matrix-org/dendrite/setup/jetstream"
@@ -29,10 +34,6 @@ import (
 	"github.com/matrix-org/dendrite/syncapi/producers"
 	"github.com/matrix-org/dendrite/syncapi/storage"
 	"github.com/matrix-org/dendrite/syncapi/types"
-	"github.com/matrix-org/gomatrixserverlib"
-	"github.com/nats-io/nats.go"
-	"github.com/sirupsen/logrus"
-	log "github.com/sirupsen/logrus"
 )
 
 // OutputClientDataConsumer consumes events that originated in the client API server.
@@ -107,7 +108,8 @@ func (s *OutputClientDataConsumer) onMessage(ctx context.Context, msg *nats.Msg)
 			"type":       output.Type,
 			"room_id":    output.RoomID,
 			log.ErrorKey: err,
-		}).Panicf("could not save account data")
+		}).Errorf("could not save account data")
+		return false
 	}
 
 	if err = s.sendReadUpdate(ctx, userID, output); err != nil {

syncapi/routing/filter.go

@@ -16,7 +16,7 @@ package routing
 
 import (
 	"encoding/json"
-	"io/ioutil"
+	"io"
 	"net/http"
 
 	"github.com/matrix-org/gomatrixserverlib"
@@ -88,7 +88,7 @@ func PutFilter(
 	var filter gomatrixserverlib.Filter
 
 	defer req.Body.Close() // nolint:errcheck
-	body, err := ioutil.ReadAll(req.Body)
+	body, err := io.ReadAll(req.Body)
 	if err != nil {
 		return util.JSONResponse{
 			Code: http.StatusBadRequest,

syncapi/routing/messages.go

@@ -20,6 +20,10 @@ import (
 	"net/http"
 	"sort"
 
+	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/matrix-org/util"
+	"github.com/sirupsen/logrus"
+
 	"github.com/matrix-org/dendrite/clientapi/jsonerror"
 	"github.com/matrix-org/dendrite/internal/caching"
 	"github.com/matrix-org/dendrite/roomserver/api"
@@ -28,9 +32,6 @@ import (
 	"github.com/matrix-org/dendrite/syncapi/sync"
 	"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"
-	"github.com/sirupsen/logrus"
 )
 
 type messagesReq struct {
@@ -262,7 +263,7 @@ func (m *messagesResp) applyLazyLoadMembers(
 		}
 	}
 	for _, evt := range membershipToUser {
-		m.State = append(m.State, gomatrixserverlib.HeaderedToClientEvent(evt, gomatrixserverlib.FormatSync))
+		m.State = append(m.State, gomatrixserverlib.HeaderedToClientEvent(evt, gomatrixserverlib.FormatAll))
 	}
 }
 

syncapi/storage/postgres/notification_data_table.go

@@ -58,7 +58,7 @@ const upsertRoomUnreadNotificationCountsSQL = `INSERT INTO syncapi_notification_
   (user_id, room_id, notification_count, highlight_count)
   VALUES ($1, $2, $3, $4)
   ON CONFLICT (user_id, room_id)
-  DO UPDATE SET notification_count = $3, highlight_count = $4
+  DO UPDATE SET id = nextval('syncapi_notification_data_id_seq'), notification_count = $3, highlight_count = $4
   RETURNING id`
 
 const selectUserUnreadNotificationCountsSQL = `SELECT

syncapi/storage/sqlite3/notification_data_table.go

@@ -25,12 +25,14 @@ import (
 	"github.com/matrix-org/dendrite/syncapi/types"
 )
 
-func NewSqliteNotificationDataTable(db *sql.DB) (tables.NotificationData, error) {
+func NewSqliteNotificationDataTable(db *sql.DB, streamID *StreamIDStatements) (tables.NotificationData, error) {
 	_, err := db.Exec(notificationDataSchema)
 	if err != nil {
 		return nil, err
 	}
-	r := &notificationDataStatements{}
+	r := &notificationDataStatements{
+		streamIDStatements: streamID,
+	}
 	return r, sqlutil.StatementList{
 		{&r.upsertRoomUnreadCounts, upsertRoomUnreadNotificationCountsSQL},
 		{&r.selectUserUnreadCounts, selectUserUnreadNotificationCountsSQL},
@@ -39,6 +41,7 @@ func NewSqliteNotificationDataTable(db *sql.DB) (tables.NotificationData, error)
 }
 
 type notificationDataStatements struct {
+	streamIDStatements     *StreamIDStatements
 	upsertRoomUnreadCounts *sql.Stmt
 	selectUserUnreadCounts *sql.Stmt
 	selectMaxID            *sql.Stmt
@@ -58,8 +61,7 @@ const upsertRoomUnreadNotificationCountsSQL = `INSERT INTO syncapi_notification_
   (user_id, room_id, notification_count, highlight_count)
   VALUES ($1, $2, $3, $4)
   ON CONFLICT (user_id, room_id)
-  DO UPDATE SET notification_count = $3, highlight_count = $4
-  RETURNING id`
+  DO UPDATE SET id = $5, notification_count = $6, highlight_count = $7`
 
 const selectUserUnreadNotificationCountsSQL = `SELECT
   id, room_id, notification_count, highlight_count
@@ -71,7 +73,11 @@ const selectUserUnreadNotificationCountsSQL = `SELECT
 const selectMaxNotificationIDSQL = `SELECT CASE COUNT(*) WHEN 0 THEN 0 ELSE MAX(id) END FROM syncapi_notification_data`
 
 func (r *notificationDataStatements) UpsertRoomUnreadCounts(ctx context.Context, userID, roomID string, notificationCount, highlightCount int) (pos types.StreamPosition, err error) {
-	err = r.upsertRoomUnreadCounts.QueryRowContext(ctx, userID, roomID, notificationCount, highlightCount).Scan(&pos)
+	pos, err = r.streamIDStatements.nextNotificationID(ctx, nil)
+	if err != nil {
+		return
+	}
+	_, err = r.upsertRoomUnreadCounts.ExecContext(ctx, userID, roomID, notificationCount, highlightCount, pos, notificationCount, highlightCount)
 	return
 }
 

syncapi/storage/sqlite3/stream_id_table.go

@@ -26,6 +26,8 @@ INSERT INTO syncapi_stream_id (stream_name, stream_id) VALUES ("invite", 0)
   ON CONFLICT DO NOTHING;
 INSERT INTO syncapi_stream_id (stream_name, stream_id) VALUES ("presence", 0)
   ON CONFLICT DO NOTHING;
+INSERT INTO syncapi_stream_id (stream_name, stream_id) VALUES ("notification", 0)
+  ON CONFLICT DO NOTHING;
 `
 
 const increaseStreamIDStmt = "" +
@@ -78,3 +80,9 @@ func (s *StreamIDStatements) nextPresenceID(ctx context.Context, txn *sql.Tx) (p
 	err = increaseStmt.QueryRowContext(ctx, "presence").Scan(&pos)
 	return
 }
+
+func (s *StreamIDStatements) nextNotificationID(ctx context.Context, txn *sql.Tx) (pos types.StreamPosition, err error) {
+	increaseStmt := sqlutil.TxStmt(txn, s.increaseStreamIDStmt)
+	err = increaseStmt.QueryRowContext(ctx, "notification").Scan(&pos)
+	return
+}

syncapi/storage/sqlite3/syncserver.go

@@ -95,7 +95,7 @@ func (d *SyncServerDatasource) prepare() (err error) {
 	if err != nil {
 		return err
 	}
-	notificationData, err := NewSqliteNotificationDataTable(d.db)
+	notificationData, err := NewSqliteNotificationDataTable(d.db, &d.streamID)
 	if err != nil {
 		return err
 	}

syncapi/streams/stream_pdu.go

@@ -12,9 +12,12 @@ import (
 	roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
 	"github.com/matrix-org/dendrite/syncapi/types"
 	userapi "github.com/matrix-org/dendrite/userapi/api"
+
 	"github.com/matrix-org/gomatrixserverlib"
 	"github.com/tidwall/gjson"
 	"go.uber.org/atomic"
+
+	"github.com/matrix-org/dendrite/syncapi/notifier"
 )
 
 // The max number of per-room goroutines to have running.
@@ -34,6 +37,7 @@ type PDUStreamProvider struct {
 	// userID+deviceID -> lazy loading cache
 	lazyLoadCache caching.LazyLoadCache
 	rsAPI         roomserverAPI.SyncRoomserverAPI
+	notifier      *notifier.Notifier
 }
 
 func (p *PDUStreamProvider) worker() {
@@ -100,6 +104,15 @@ func (p *PDUStreamProvider) CompleteSync(
 		req.Log.WithError(err).Error("unable to update event filter with ignored users")
 	}
 
+	// Invalidate the lazyLoadCache, otherwise we end up with missing displaynames/avatars
+	// TODO: This might be inefficient, when joined to many and/or large rooms.
+	for _, roomID := range joinedRoomIDs {
+		joinedUsers := p.notifier.JoinedUsers(roomID)
+		for _, sharedUser := range joinedUsers {
+			p.lazyLoadCache.InvalidateLazyLoadedUser(req.Device, roomID, sharedUser)
+		}
+	}
+
 	// Build up a /sync response. Add joined rooms.
 	var reqMutex sync.Mutex
 	var reqWaitGroup sync.WaitGroup

syncapi/streams/streams.go

@@ -34,6 +34,7 @@ func NewSyncStreamProviders(
 			StreamProvider: StreamProvider{DB: d},
 			lazyLoadCache:  lazyLoadCache,
 			rsAPI:          rsAPI,
+			notifier:       notifier,
 		},
 		TypingStreamProvider: &TypingStreamProvider{
 			StreamProvider: StreamProvider{DB: d},

test/keys.go

@@ -22,7 +22,6 @@ import (
 	"encoding/pem"
 	"errors"
 	"fmt"
-	"io/ioutil"
 	"math/big"
 	"os"
 	"strings"
@@ -144,7 +143,7 @@ func NewTLSKeyWithAuthority(serverName, tlsKeyPath, tlsCertPath, authorityKeyPat
 	}
 
 	// load the authority key
-	dat, err := ioutil.ReadFile(authorityKeyPath)
+	dat, err := os.ReadFile(authorityKeyPath)
 	if err != nil {
 		return err
 	}
@@ -158,7 +157,7 @@ func NewTLSKeyWithAuthority(serverName, tlsKeyPath, tlsCertPath, authorityKeyPat
 	}
 
 	// load the authority certificate
-	dat, err = ioutil.ReadFile(authorityCertPath)
+	dat, err = os.ReadFile(authorityCertPath)
 	if err != nil {
 		return err
 	}