Merge branch 'master' into neilalexander/lessevents

federationsender/storage/postgres/deltas/2021020411080000_rooms.go

@@ -0,0 +1,46 @@
+// Copyright 2021 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 deltas
+
+import (
+	"database/sql"
+	"fmt"
+
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+	"github.com/pressly/goose"
+)
+
+func LoadFromGoose() {
+	goose.AddMigration(UpRemoveRoomsTable, DownRemoveRoomsTable)
+}
+
+func LoadRemoveRoomsTable(m *sqlutil.Migrations) {
+	m.AddMigration(UpRemoveRoomsTable, DownRemoveRoomsTable)
+}
+
+func UpRemoveRoomsTable(tx *sql.Tx) error {
+	_, err := tx.Exec(`
+		DROP TABLE IF EXISTS federationsender_rooms;
+	`)
+	if err != nil {
+		return fmt.Errorf("failed to execute upgrade: %w", err)
+	}
+	return nil
+}
+
+func DownRemoveRoomsTable(tx *sql.Tx) error {
+	// We can't reverse this.
+	return nil
+}

federationsender/storage/postgres/room_table.go

@@ -1,104 +0,0 @@
-// Copyright 2017-2018 New Vector Ltd
-// Copyright 2019-2020 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 postgres
-
-import (
-	"context"
-	"database/sql"
-
-	"github.com/matrix-org/dendrite/internal/sqlutil"
-)
-
-const roomSchema = `
-CREATE TABLE IF NOT EXISTS federationsender_rooms (
-    -- The string ID of the room
-    room_id TEXT PRIMARY KEY,
-    -- The most recent event state by the room server.
-    -- We can use this to tell if our view of the room state has become
-    -- desynchronised.
-    last_event_id TEXT NOT NULL
-);`
-
-const insertRoomSQL = "" +
-	"INSERT INTO federationsender_rooms (room_id, last_event_id) VALUES ($1, '')" +
-	" ON CONFLICT DO NOTHING"
-
-const selectRoomForUpdateSQL = "" +
-	"SELECT last_event_id FROM federationsender_rooms WHERE room_id = $1 FOR UPDATE"
-
-const updateRoomSQL = "" +
-	"UPDATE federationsender_rooms SET last_event_id = $2 WHERE room_id = $1"
-
-type roomStatements struct {
-	db                      *sql.DB
-	insertRoomStmt          *sql.Stmt
-	selectRoomForUpdateStmt *sql.Stmt
-	updateRoomStmt          *sql.Stmt
-}
-
-func NewPostgresRoomsTable(db *sql.DB) (s *roomStatements, err error) {
-	s = &roomStatements{
-		db: db,
-	}
-	_, err = s.db.Exec(roomSchema)
-	if err != nil {
-		return
-	}
-	if s.insertRoomStmt, err = s.db.Prepare(insertRoomSQL); err != nil {
-		return
-	}
-	if s.selectRoomForUpdateStmt, err = s.db.Prepare(selectRoomForUpdateSQL); err != nil {
-		return
-	}
-	if s.updateRoomStmt, err = s.db.Prepare(updateRoomSQL); err != nil {
-		return
-	}
-	return
-}
-
-// insertRoom inserts the room if it didn't already exist.
-// If the room didn't exist then last_event_id is set to the empty string.
-func (s *roomStatements) InsertRoom(
-	ctx context.Context, txn *sql.Tx, roomID string,
-) error {
-	_, err := sqlutil.TxStmt(txn, s.insertRoomStmt).ExecContext(ctx, roomID)
-	return err
-}
-
-// selectRoomForUpdate locks the row for the room and returns the last_event_id.
-// The row must already exist in the table. Callers can ensure that the row
-// exists by calling insertRoom first.
-func (s *roomStatements) SelectRoomForUpdate(
-	ctx context.Context, txn *sql.Tx, roomID string,
-) (string, error) {
-	var lastEventID string
-	stmt := sqlutil.TxStmt(txn, s.selectRoomForUpdateStmt)
-	err := stmt.QueryRowContext(ctx, roomID).Scan(&lastEventID)
-	if err != nil {
-		return "", err
-	}
-	return lastEventID, nil
-}
-
-// updateRoom updates the last_event_id for the room. selectRoomForUpdate should
-// have already been called earlier within the transaction.
-func (s *roomStatements) UpdateRoom(
-	ctx context.Context, txn *sql.Tx, roomID, lastEventID string,
-) error {
-	stmt := sqlutil.TxStmt(txn, s.updateRoomStmt)
-	_, err := stmt.ExecContext(ctx, roomID, lastEventID)
-	return err
-}

federationsender/storage/postgres/storage.go

@@ -18,6 +18,7 @@ package postgres
 import (
 	"database/sql"
 
+	"github.com/matrix-org/dendrite/federationsender/storage/postgres/deltas"
 	"github.com/matrix-org/dendrite/federationsender/storage/shared"
 	"github.com/matrix-org/dendrite/internal/caching"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
@@ -56,10 +57,6 @@ func NewDatabase(dbProperties *config.DatabaseOptions, cache caching.FederationS
 	if err != nil {
 		return nil, err
 	}
-	rooms, err := NewPostgresRoomsTable(d.db)
-	if err != nil {
-		return nil, err
-	}
 	blacklist, err := NewPostgresBlacklistTable(d.db)
 	if err != nil {
 		return nil, err
@@ -72,6 +69,11 @@ func NewDatabase(dbProperties *config.DatabaseOptions, cache caching.FederationS
 	if err != nil {
 		return nil, err
 	}
+	m := sqlutil.NewMigrations()
+	deltas.LoadRemoveRoomsTable(m)
+	if err = m.RunDeltas(d.db, dbProperties); err != nil {
+		return nil, err
+	}
 	d.Database = shared.Database{
 		DB:                            d.db,
 		Cache:                         cache,
@@ -80,7 +82,6 @@ func NewDatabase(dbProperties *config.DatabaseOptions, cache caching.FederationS
 		FederationSenderQueuePDUs:     queuePDUs,
 		FederationSenderQueueEDUs:     queueEDUs,
 		FederationSenderQueueJSON:     queueJSON,
-		FederationSenderRooms:         rooms,
 		FederationSenderBlacklist:     blacklist,
 		FederationSenderInboundPeeks:  inboundPeeks,
 		FederationSenderOutboundPeeks: outboundPeeks,

federationsender/storage/shared/storage.go

@@ -34,7 +34,6 @@ type Database struct {
 	FederationSenderQueueEDUs     tables.FederationSenderQueueEDUs
 	FederationSenderQueueJSON     tables.FederationSenderQueueJSON
 	FederationSenderJoinedHosts   tables.FederationSenderJoinedHosts
-	FederationSenderRooms         tables.FederationSenderRooms
 	FederationSenderBlacklist     tables.FederationSenderBlacklist
 	FederationSenderOutboundPeeks tables.FederationSenderOutboundPeeks
 	FederationSenderInboundPeeks  tables.FederationSenderInboundPeeks
@@ -64,29 +63,6 @@ func (d *Database) UpdateRoom(
 	removeHosts []string,
 ) (joinedHosts []types.JoinedHost, err error) {
 	err = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
-		err = d.FederationSenderRooms.InsertRoom(ctx, txn, roomID)
-		if err != nil {
-			return err
-		}
-
-		lastSentEventID, err := d.FederationSenderRooms.SelectRoomForUpdate(ctx, txn, roomID)
-		if err != nil {
-			return err
-		}
-
-		if lastSentEventID == newEventID {
-			// We've handled this message before, so let's just ignore it.
-			// We can only get a duplicate for the last message we processed,
-			// so its enough just to compare the newEventID with lastSentEventID
-			return nil
-		}
-
-		if lastSentEventID != "" && lastSentEventID != oldEventID {
-			return types.EventIDMismatchError{
-				DatabaseID: lastSentEventID, RoomServerID: oldEventID,
-			}
-		}
-
 		joinedHosts, err = d.FederationSenderJoinedHosts.SelectJoinedHostsWithTx(ctx, txn, roomID)
 		if err != nil {
 			return err
@@ -101,7 +77,7 @@ func (d *Database) UpdateRoom(
 		if err = d.FederationSenderJoinedHosts.DeleteJoinedHosts(ctx, txn, removeHosts); err != nil {
 			return err
 		}
-		return d.FederationSenderRooms.UpdateRoom(ctx, txn, roomID, newEventID)
+		return nil
 	})
 	return
 }

federationsender/storage/sqlite3/deltas/2021020411080000_rooms.go

@@ -0,0 +1,46 @@
+// Copyright 2021 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 deltas
+
+import (
+	"database/sql"
+	"fmt"
+
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+	"github.com/pressly/goose"
+)
+
+func LoadFromGoose() {
+	goose.AddMigration(UpRemoveRoomsTable, DownRemoveRoomsTable)
+}
+
+func LoadRemoveRoomsTable(m *sqlutil.Migrations) {
+	m.AddMigration(UpRemoveRoomsTable, DownRemoveRoomsTable)
+}
+
+func UpRemoveRoomsTable(tx *sql.Tx) error {
+	_, err := tx.Exec(`
+		DROP TABLE IF EXISTS federationsender_rooms;
+	`)
+	if err != nil {
+		return fmt.Errorf("failed to execute upgrade: %w", err)
+	}
+	return nil
+}
+
+func DownRemoveRoomsTable(tx *sql.Tx) error {
+	// We can't reverse this.
+	return nil
+}

federationsender/storage/sqlite3/room_table.go

@@ -1,105 +0,0 @@
-// Copyright 2017-2018 New Vector Ltd
-// Copyright 2019-2020 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 sqlite3
-
-import (
-	"context"
-	"database/sql"
-
-	"github.com/matrix-org/dendrite/internal/sqlutil"
-)
-
-const roomSchema = `
-CREATE TABLE IF NOT EXISTS federationsender_rooms (
-    -- The string ID of the room
-    room_id TEXT PRIMARY KEY,
-    -- The most recent event state by the room server.
-    -- We can use this to tell if our view of the room state has become
-    -- desynchronised.
-    last_event_id TEXT NOT NULL
-);`
-
-const insertRoomSQL = "" +
-	"INSERT INTO federationsender_rooms (room_id, last_event_id) VALUES ($1, '')" +
-	" ON CONFLICT DO NOTHING"
-
-const selectRoomForUpdateSQL = "" +
-	"SELECT last_event_id FROM federationsender_rooms WHERE room_id = $1"
-
-const updateRoomSQL = "" +
-	"UPDATE federationsender_rooms SET last_event_id = $2 WHERE room_id = $1"
-
-type roomStatements struct {
-	db                      *sql.DB
-	insertRoomStmt          *sql.Stmt
-	selectRoomForUpdateStmt *sql.Stmt
-	updateRoomStmt          *sql.Stmt
-}
-
-func NewSQLiteRoomsTable(db *sql.DB) (s *roomStatements, err error) {
-	s = &roomStatements{
-		db: db,
-	}
-	_, err = db.Exec(roomSchema)
-	if err != nil {
-		return
-	}
-
-	if s.insertRoomStmt, err = db.Prepare(insertRoomSQL); err != nil {
-		return
-	}
-	if s.selectRoomForUpdateStmt, err = db.Prepare(selectRoomForUpdateSQL); err != nil {
-		return
-	}
-	if s.updateRoomStmt, err = db.Prepare(updateRoomSQL); err != nil {
-		return
-	}
-	return
-}
-
-// insertRoom inserts the room if it didn't already exist.
-// If the room didn't exist then last_event_id is set to the empty string.
-func (s *roomStatements) InsertRoom(
-	ctx context.Context, txn *sql.Tx, roomID string,
-) error {
-	_, err := sqlutil.TxStmt(txn, s.insertRoomStmt).ExecContext(ctx, roomID)
-	return err
-}
-
-// selectRoomForUpdate locks the row for the room and returns the last_event_id.
-// The row must already exist in the table. Callers can ensure that the row
-// exists by calling insertRoom first.
-func (s *roomStatements) SelectRoomForUpdate(
-	ctx context.Context, txn *sql.Tx, roomID string,
-) (string, error) {
-	var lastEventID string
-	stmt := sqlutil.TxStmt(txn, s.selectRoomForUpdateStmt)
-	err := stmt.QueryRowContext(ctx, roomID).Scan(&lastEventID)
-	if err != nil {
-		return "", err
-	}
-	return lastEventID, nil
-}
-
-// updateRoom updates the last_event_id for the room. selectRoomForUpdate should
-// have already been called earlier within the transaction.
-func (s *roomStatements) UpdateRoom(
-	ctx context.Context, txn *sql.Tx, roomID, lastEventID string,
-) error {
-	stmt := sqlutil.TxStmt(txn, s.updateRoomStmt)
-	_, err := stmt.ExecContext(ctx, roomID, lastEventID)
-	return err
-}

federationsender/storage/sqlite3/storage.go

@@ -21,6 +21,7 @@ import (
 	_ "github.com/mattn/go-sqlite3"
 
 	"github.com/matrix-org/dendrite/federationsender/storage/shared"
+	"github.com/matrix-org/dendrite/federationsender/storage/sqlite3/deltas"
 	"github.com/matrix-org/dendrite/internal/caching"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/setup/config"
@@ -46,10 +47,6 @@ func NewDatabase(dbProperties *config.DatabaseOptions, cache caching.FederationS
 	if err != nil {
 		return nil, err
 	}
-	rooms, err := NewSQLiteRoomsTable(d.db)
-	if err != nil {
-		return nil, err
-	}
 	queuePDUs, err := NewSQLiteQueuePDUsTable(d.db)
 	if err != nil {
 		return nil, err
@@ -74,6 +71,11 @@ func NewDatabase(dbProperties *config.DatabaseOptions, cache caching.FederationS
 	if err != nil {
 		return nil, err
 	}
+	m := sqlutil.NewMigrations()
+	deltas.LoadRemoveRoomsTable(m)
+	if err = m.RunDeltas(d.db, dbProperties); err != nil {
+		return nil, err
+	}
 	d.Database = shared.Database{
 		DB:                            d.db,
 		Cache:                         cache,
@@ -82,7 +84,6 @@ func NewDatabase(dbProperties *config.DatabaseOptions, cache caching.FederationS
 		FederationSenderQueuePDUs:     queuePDUs,
 		FederationSenderQueueEDUs:     queueEDUs,
 		FederationSenderQueueJSON:     queueJSON,
-		FederationSenderRooms:         rooms,
 		FederationSenderBlacklist:     blacklist,
 		FederationSenderOutboundPeeks: outboundPeeks,
 		FederationSenderInboundPeeks:  inboundPeeks,

federationsender/storage/tables/interface.go

@@ -56,12 +56,6 @@ type FederationSenderJoinedHosts interface {
 	SelectJoinedHostsForRooms(ctx context.Context, roomIDs []string) ([]gomatrixserverlib.ServerName, error)
 }
 
-type FederationSenderRooms interface {
-	InsertRoom(ctx context.Context, txn *sql.Tx, roomID string) error
-	SelectRoomForUpdate(ctx context.Context, txn *sql.Tx, roomID string) (string, error)
-	UpdateRoom(ctx context.Context, txn *sql.Tx, roomID, lastEventID string) error
-}
-
 type FederationSenderBlacklist interface {
 	InsertBlacklist(ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName) error
 	SelectBlacklist(ctx context.Context, txn *sql.Tx, serverName gomatrixserverlib.ServerName) (bool, error)

federationsender/types/types.go

@@ -15,8 +15,6 @@
 package types
 
 import (
-	"fmt"
-
 	"github.com/matrix-org/gomatrixserverlib"
 )
 
@@ -34,22 +32,6 @@ func (s ServerNames) Len() int           { return len(s) }
 func (s ServerNames) Swap(i, j int)      { s[i], s[j] = s[j], s[i] }
 func (s ServerNames) Less(i, j int) bool { return s[i] < s[j] }
 
-// A EventIDMismatchError indicates that we have got out of sync with the
-// room server.
-type EventIDMismatchError struct {
-	// The event ID we have stored in our local database.
-	DatabaseID string
-	// The event ID received from the room server.
-	RoomServerID string
-}
-
-func (e EventIDMismatchError) Error() string {
-	return fmt.Sprintf(
-		"mismatched last sent event ID: had %q in database got %q from room server",
-		e.DatabaseID, e.RoomServerID,
-	)
-}
-
 // tracks peeks we're performing on another server over federation
 type OutboundPeek struct {
 	PeekID            string

go.mod

@@ -22,7 +22,7 @@ require (
 	github.com/matrix-org/go-http-js-libp2p v0.0.0-20200518170932-783164aeeda4
 	github.com/matrix-org/go-sqlite3-js v0.0.0-20200522092705-bc8506ccbcf3
 	github.com/matrix-org/gomatrix v0.0.0-20200827122206-7dd5e2a05bcd
-	github.com/matrix-org/gomatrixserverlib v0.0.0-20210128131744-c803b6ee2b68
+	github.com/matrix-org/gomatrixserverlib v0.0.0-20210129163316-dd4d53729ead
 	github.com/matrix-org/naffka v0.0.0-20200901083833-bcdd62999a91
 	github.com/matrix-org/util v0.0.0-20200807132607-55161520e1d4
 	github.com/mattn/go-sqlite3 v1.14.2

go.sum

@@ -567,8 +567,8 @@ github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26 h1:Hr3zjRsq2bh
 github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26/go.mod h1:3fxX6gUjWyI/2Bt7J1OLhpCzOfO/bB3AiX0cJtEKud0=
 github.com/matrix-org/gomatrix v0.0.0-20200827122206-7dd5e2a05bcd h1:xVrqJK3xHREMNjwjljkAUaadalWc0rRbmVuQatzmgwg=
 github.com/matrix-org/gomatrix v0.0.0-20200827122206-7dd5e2a05bcd/go.mod h1:/gBX06Kw0exX1HrwmoBibFA98yBk/jxKpGVeyQbff+s=
-github.com/matrix-org/gomatrixserverlib v0.0.0-20210128131744-c803b6ee2b68 h1:zZouhQEqn2J0eiRlFwkANVmLXTltpCanEmDdJyPIOhk=
+github.com/matrix-org/gomatrixserverlib v0.0.0-20210129163316-dd4d53729ead h1:VmGJybKUQin8+NyA9ZkrHJpE8ygXzcON9peQH9LC92c=
-github.com/matrix-org/gomatrixserverlib v0.0.0-20210128131744-c803b6ee2b68/go.mod h1:JsAzE1Ll3+gDWS9JSUHPJiiyAksvOOnGWF2nXdg4ZzU=
+github.com/matrix-org/gomatrixserverlib v0.0.0-20210129163316-dd4d53729ead/go.mod h1:JsAzE1Ll3+gDWS9JSUHPJiiyAksvOOnGWF2nXdg4ZzU=
 github.com/matrix-org/naffka v0.0.0-20200901083833-bcdd62999a91 h1:HJ6U3S3ljJqNffYMcIeAncp5qT/i+ZMiJ2JC2F0aXP4=
 github.com/matrix-org/naffka v0.0.0-20200901083833-bcdd62999a91/go.mod h1:sjyPyRxKM5uw1nD2cJ6O2OxI6GOqyVBfNXqKjBZTBZE=
 github.com/matrix-org/util v0.0.0-20190711121626-527ce5ddefc7 h1:ntrLa/8xVzeSs8vHFHK25k0C+NV74sYMJnNSg5NoSRo=

roomserver/internal/input/input_membership.go

@@ -107,13 +107,6 @@ func (r *Inputer) updateMembership(
 		return updates, nil
 	}
 
-	if add == nil {
-		// This can happen when we have rejoined a room and suddenly we have a
-		// divergence between the former state and the new one. We don't want to
-		// act on removals and apparently there are no adds, so stop here.
-		return updates, nil
-	}
-
 	mu, err := updater.MembershipUpdater(targetUserNID, r.isLocalTarget(add))
 	if err != nil {
 		return nil, err

roomserver/internal/perform/perform_invite.go

@@ -225,7 +225,7 @@ func buildInviteStrippedState(
 	for _, t := range []string{
 		gomatrixserverlib.MRoomName, gomatrixserverlib.MRoomCanonicalAlias,
 		gomatrixserverlib.MRoomAliases, gomatrixserverlib.MRoomJoinRules,
-		"m.room.avatar", "m.room.encryption",
+		"m.room.avatar", "m.room.encryption", gomatrixserverlib.MRoomCreate,
 	} {
 		stateWanted = append(stateWanted, gomatrixserverlib.StateKeyTuple{
 			EventType: t,

syncapi/storage/postgres/account_data_table.go

@@ -53,7 +53,7 @@ CREATE UNIQUE INDEX IF NOT EXISTS syncapi_account_data_id_idx ON syncapi_account
 const insertAccountDataSQL = "" +
 	"INSERT INTO syncapi_account_data_type (user_id, room_id, type) VALUES ($1, $2, $3)" +
 	" ON CONFLICT ON CONSTRAINT syncapi_account_data_unique" +
-	" DO UPDATE SET id = EXCLUDED.id" +
+	" DO UPDATE SET id = nextval('syncapi_stream_id')" +
 	" RETURNING id"
 
 const selectAccountDataInRangeSQL = "" +

syncapi/storage/sqlite3/account_data_table.go

@@ -39,7 +39,7 @@ CREATE TABLE IF NOT EXISTS syncapi_account_data_type (
 const insertAccountDataSQL = "" +
 	"INSERT INTO syncapi_account_data_type (id, user_id, room_id, type) VALUES ($1, $2, $3, $4)" +
 	" ON CONFLICT (user_id, room_id, type) DO UPDATE" +
-	" SET id = EXCLUDED.id"
+	" SET id = $5"
 
 const selectAccountDataInRangeSQL = "" +
 	"SELECT room_id, type FROM syncapi_account_data_type" +
@@ -86,7 +86,7 @@ func (s *accountDataStatements) InsertAccountData(
 	if err != nil {
 		return
 	}
-	_, err = sqlutil.TxStmt(txn, s.insertAccountDataStmt).ExecContext(ctx, pos, userID, roomID, dataType)
+	_, err = sqlutil.TxStmt(txn, s.insertAccountDataStmt).ExecContext(ctx, pos, userID, roomID, dataType, pos)
 	return
 }
 

syncapi/streams/stream_devicelist.go

@@ -19,7 +19,7 @@ func (p *DeviceListStreamProvider) CompleteSync(
 	ctx context.Context,
 	req *types.SyncRequest,
 ) types.LogPosition {
-	return p.IncrementalSync(ctx, req, types.LogPosition{}, p.LatestPosition(ctx))
+	return p.LatestPosition(ctx)
 }
 
 func (p *DeviceListStreamProvider) IncrementalSync(

syncapi/streams/stream_pdu.go

@@ -2,18 +2,54 @@ package streams
 
 import (
 	"context"
+	"sync"
+	"time"
 
 	"github.com/matrix-org/dendrite/syncapi/types"
 	userapi "github.com/matrix-org/dendrite/userapi/api"
 	"github.com/matrix-org/gomatrixserverlib"
+	"go.uber.org/atomic"
 )
 
+// The max number of per-room goroutines to have running.
+// Too high and this will consume lots of CPU, too low and complete
+// sync responses will take longer to process.
+const PDU_STREAM_WORKERS = 256
+
+// The maximum number of tasks that can be queued in total before
+// backpressure will build up and the rests will start to block.
+const PDU_STREAM_QUEUESIZE = PDU_STREAM_WORKERS * 8
+
 type PDUStreamProvider struct {
 	StreamProvider
+
+	tasks   chan func()
+	workers atomic.Int32
+}
+
+func (p *PDUStreamProvider) worker() {
+	defer p.workers.Dec()
+	for {
+		select {
+		case f := <-p.tasks:
+			f()
+		case <-time.After(time.Second * 10):
+			return
+		}
+	}
+}
+
+func (p *PDUStreamProvider) queue(f func()) {
+	if p.workers.Load() < PDU_STREAM_WORKERS {
+		p.workers.Inc()
+		go p.worker()
+	}
+	p.tasks <- f
 }
 
 func (p *PDUStreamProvider) Setup() {
 	p.StreamProvider.Setup()
+	p.tasks = make(chan func(), PDU_STREAM_QUEUESIZE)
 
 	p.latestMutex.Lock()
 	defer p.latestMutex.Unlock()
@@ -52,19 +88,32 @@ func (p *PDUStreamProvider) CompleteSync(
 	eventFilter := req.Filter.Room.Timeline
 
 	// Build up a /sync response. Add joined rooms.
-	for _, roomID := range joinedRoomIDs {
+	var reqMutex sync.Mutex
-		var jr *types.JoinResponse
+	var reqWaitGroup sync.WaitGroup
-		jr, err = p.getJoinResponseForCompleteSync(
+	reqWaitGroup.Add(len(joinedRoomIDs))
-			ctx, roomID, r, &stateFilter, &eventFilter, req.WantFullState, req.Device,
+	for _, room := range joinedRoomIDs {
-		)
+		roomID := room
-		if err != nil {
+		p.queue(func() {
-			req.Log.WithError(err).Error("p.getJoinResponseForCompleteSync failed")
+			defer reqWaitGroup.Done()
-			return from
+
-		}
+			var jr *types.JoinResponse
-		req.Response.Rooms.Join[roomID] = *jr
+			jr, err = p.getJoinResponseForCompleteSync(
-		req.Rooms[roomID] = gomatrixserverlib.Join
+				ctx, roomID, r, &stateFilter, &eventFilter, req.WantFullState, req.Device,
+			)
+			if err != nil {
+				req.Log.WithError(err).Error("p.getJoinResponseForCompleteSync failed")
+				return
+			}
+
+			reqMutex.Lock()
+			defer reqMutex.Unlock()
+			req.Response.Rooms.Join[roomID] = *jr
+			req.Rooms[roomID] = gomatrixserverlib.Join
+		})
 	}
 
+	reqWaitGroup.Wait()
+
 	// Add peeked rooms.
 	peeks, err := p.DB.PeeksInRange(ctx, req.Device.UserID, req.Device.ID, r)
 	if err != nil {