Merge branch 'master' into neilalexander/onoldroomevent

CHANGES.md

@@ -1,5 +1,19 @@
 # Changelog
 
+## Dendrite 0.3.9 (2021-02-04)
+
+### Features
+
+* Performance of initial/complete syncs has been improved dramatically
+* State events that can't be authed are now dropped when joining a room rather than unexpectedly causing the room join to fail
+* State events that already appear in the timeline will no longer be requested from the sync API database more than once, which may reduce memory usage in some cases
+
+### Fixes
+
+* A crash at startup due to a conflict in the sync API account data has been fixed
+* A crash at startup due to mismatched event IDs in the federation sender has been fixed
+* A redundant check which may cause the roomserver memberships table to get out of sync has been removed
+
 ## Dendrite 0.3.8 (2021-01-28)
 
 ### Fixes

build/docker/docker-compose.polylith.yml

@@ -70,7 +70,7 @@ services:
     volumes:
       - ./config:/etc/dendrite
     networks:
-        - internal
+      - internal
 
   signing_key_server:
     hostname: signing_key_server
@@ -86,9 +86,9 @@ services:
     image: matrixdotorg/dendrite-polylith:latest
     command: userapi
     volumes:
-        - ./config:/etc/dendrite
+      - ./config:/etc/dendrite
     networks:
-        - internal
+      - internal
 
   appservice_api:
     hostname: appservice_api

federationsender/queue/destinationqueue.go

@@ -444,7 +444,7 @@ func (oq *destinationQueue) nextTransaction(
 		log.WithFields(log.Fields{
 			"destination": oq.destination,
 			log.ErrorKey:  err,
-		}).Infof("Failed to send transaction %q", t.TransactionID)
+		}).Debugf("Failed to send transaction %q", t.TransactionID)
 		return false, 0, 0, err
 	}
 }

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

internal/version.go

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

roomserver/internal/input/input_membership.go

@@ -107,6 +107,23 @@ func (r *Inputer) updateMembership(
 		return updates, nil
 	}
 
+	// In an ideal world, we shouldn't ever have "add" be nil and "remove" be
+	// set, as this implies that we're deleting a state event without replacing
+	// it (a thing that ordinarily shouldn't happen in Matrix). However, state
+	// resets are sadly a thing occasionally and we have to account for that.
+	// Beforehand there used to be a check here which stopped dead if we hit
+	// this scenario, but that meant that the membership table got out of sync
+	// after a state reset, often thinking that the user was still joined to
+	// the room even though the room state said otherwise, and this would prevent
+	// the user from being able to attempt to rejoin the room without modifying
+	// the database. So instead what we'll do is we'll just update the membership
+	// table to say that the user is "leave" and we'll use the old event to
+	// avoid nil pointer exceptions on the code path that follows.
+	if add == nil {
+		add = remove
+		newMembership = gomatrixserverlib.Leave
+	}
+
 	mu, err := updater.MembershipUpdater(targetUserNID, r.isLocalTarget(add))
 	if err != nil {
 		return nil, err

setup/mscs/msc2946/msc2946.go

@@ -46,7 +46,7 @@ const (
 
 // Defaults sets the request defaults
 func Defaults(r *gomatrixserverlib.MSC2946SpacesRequest) {
-	r.Limit = 100
+	r.Limit = 2000
 	r.MaxRoomsPerSpace = -1
 }
 
@@ -108,9 +108,6 @@ func federatedSpacesHandler(
 			JSON: jsonerror.BadJSON("The request body could not be decoded into valid JSON. " + err.Error()),
 		}
 	}
-	if r.Limit > 100 {
-		r.Limit = 100
-	}
 	w := walker{
 		req:        &r,
 		rootRoomID: roomID,
@@ -147,9 +144,6 @@ func spacesHandler(
 		if resErr := chttputil.UnmarshalJSONRequest(req, &r); resErr != nil {
 			return *resErr
 		}
-		if r.Limit > 100 {
-			r.Limit = 100
-		}
 		w := walker{
 			req:        &r,
 			rootRoomID: roomID,

syncapi/storage/interface.go

@@ -35,7 +35,7 @@ type Database interface {
 	MaxStreamPositionForAccountData(ctx context.Context) (types.StreamPosition, error)
 	MaxStreamPositionForSendToDeviceMessages(ctx context.Context) (types.StreamPosition, error)
 
-	CurrentState(ctx context.Context, roomID string, stateFilterPart *gomatrixserverlib.StateFilter) ([]*gomatrixserverlib.HeaderedEvent, error)
+	CurrentState(ctx context.Context, roomID string, stateFilterPart *gomatrixserverlib.StateFilter, excludeEventIDs []string) ([]*gomatrixserverlib.HeaderedEvent, error)
 	GetStateDeltasForFullStateSync(ctx context.Context, device *userapi.Device, r types.Range, userID string, stateFilter *gomatrixserverlib.StateFilter) ([]types.StateDelta, []string, error)
 	GetStateDeltas(ctx context.Context, device *userapi.Device, r types.Range, userID string, stateFilter *gomatrixserverlib.StateFilter) ([]types.StateDelta, []string, error)
 	RoomIDsWithMembership(ctx context.Context, userID string, membership string) ([]string, error)

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/postgres/current_room_state_table.go

@@ -84,7 +84,8 @@ const selectCurrentStateSQL = "" +
 	" AND ( $4::text[] IS NULL OR     type LIKE ANY($4)  )" +
 	" AND ( $5::text[] IS NULL OR NOT(type LIKE ANY($5)) )" +
 	" AND ( $6::bool IS NULL   OR     contains_url = $6  )" +
-	" LIMIT $7"
+	" AND (event_id = ANY($7)) IS NOT TRUE" +
+	" LIMIT $8"
 
 const selectJoinedUsersSQL = "" +
 	"SELECT room_id, state_key FROM syncapi_current_room_state WHERE type = 'm.room.member' AND membership = 'join'"
@@ -197,6 +198,7 @@ func (s *currentRoomStateStatements) SelectRoomIDsWithMembership(
 func (s *currentRoomStateStatements) SelectCurrentState(
 	ctx context.Context, txn *sql.Tx, roomID string,
 	stateFilter *gomatrixserverlib.StateFilter,
+	excludeEventIDs []string,
 ) ([]*gomatrixserverlib.HeaderedEvent, error) {
 	stmt := sqlutil.TxStmt(txn, s.selectCurrentStateStmt)
 	rows, err := stmt.QueryContext(ctx, roomID,
@@ -205,6 +207,7 @@ func (s *currentRoomStateStatements) SelectCurrentState(
 		pq.StringArray(filterConvertTypeWildcardToSQL(stateFilter.Types)),
 		pq.StringArray(filterConvertTypeWildcardToSQL(stateFilter.NotTypes)),
 		stateFilter.ContainsURL,
+		pq.StringArray(excludeEventIDs),
 		stateFilter.Limit,
 	)
 	if err != nil {

syncapi/storage/shared/syncserver.go

@@ -103,8 +103,8 @@ func (d *Database) MaxStreamPositionForAccountData(ctx context.Context) (types.S
 	return types.StreamPosition(id), nil
 }
 
-func (d *Database) CurrentState(ctx context.Context, roomID string, stateFilterPart *gomatrixserverlib.StateFilter) ([]*gomatrixserverlib.HeaderedEvent, error) {
-	return d.CurrentRoomState.SelectCurrentState(ctx, nil, roomID, stateFilterPart)
+func (d *Database) CurrentState(ctx context.Context, roomID string, stateFilterPart *gomatrixserverlib.StateFilter, excludeEventIDs []string) ([]*gomatrixserverlib.HeaderedEvent, error) {
+	return d.CurrentRoomState.SelectCurrentState(ctx, nil, roomID, stateFilterPart, excludeEventIDs)
 }
 
 func (d *Database) RoomIDsWithMembership(ctx context.Context, userID string, membership string) ([]string, error) {
@@ -195,7 +195,7 @@ func (d *Database) GetStateEvent(
 func (d *Database) GetStateEventsForRoom(
 	ctx context.Context, roomID string, stateFilter *gomatrixserverlib.StateFilter,
 ) (stateEvents []*gomatrixserverlib.HeaderedEvent, err error) {
-	stateEvents, err = d.CurrentRoomState.SelectCurrentState(ctx, nil, roomID, stateFilter)
+	stateEvents, err = d.CurrentRoomState.SelectCurrentState(ctx, nil, roomID, stateFilter, nil)
 	return
 }
 
@@ -870,7 +870,7 @@ func (d *Database) currentStateStreamEventsForRoom(
 	ctx context.Context, txn *sql.Tx, roomID string,
 	stateFilter *gomatrixserverlib.StateFilter,
 ) ([]types.StreamEvent, error) {
-	allState, err := d.CurrentRoomState.SelectCurrentState(ctx, txn, roomID, stateFilter)
+	allState, err := d.CurrentRoomState.SelectCurrentState(ctx, txn, roomID, stateFilter, nil)
 	if err != nil {
 		return nil, err
 	}

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/storage/sqlite3/current_room_state_table.go

@@ -178,6 +178,7 @@ func (s *currentRoomStateStatements) SelectRoomIDsWithMembership(
 func (s *currentRoomStateStatements) SelectCurrentState(
 	ctx context.Context, txn *sql.Tx, roomID string,
 	stateFilter *gomatrixserverlib.StateFilter,
+	excludeEventIDs []string,
 ) ([]*gomatrixserverlib.HeaderedEvent, error) {
 	stmt, params, err := prepareWithFilters(
 		s.db, txn, selectCurrentStateSQL,
@@ -186,7 +187,7 @@ func (s *currentRoomStateStatements) SelectCurrentState(
 		},
 		stateFilter.Senders, stateFilter.NotSenders,
 		stateFilter.Types, stateFilter.NotTypes,
-		stateFilter.Limit, FilterOrderNone,
+		excludeEventIDs, stateFilter.Limit, FilterOrderNone,
 	)
 	if err != nil {
 		return nil, fmt.Errorf("s.prepareWithFilters: %w", err)

syncapi/storage/sqlite3/filtering.go

@@ -23,9 +23,10 @@ const (
 // fields might come from either a StateFilter or an EventFilter,
 // and it's easier just to have the caller extract the relevant
 // parts.
+// nolint:gocyclo
 func prepareWithFilters(
 	db *sql.DB, txn *sql.Tx, query string, params []interface{},
-	senders, notsenders, types, nottypes []string,
+	senders, notsenders, types, nottypes []string, excludeEventIDs []string,
 	limit int, order FilterOrder,
 ) (*sql.Stmt, []interface{}, error) {
 	offset := len(params)
@@ -53,6 +54,12 @@ func prepareWithFilters(
 			params, offset = append(params, v), offset+1
 		}
 	}
+	if count := len(excludeEventIDs); count > 0 {
+		query += " AND event_id NOT IN " + sqlutil.QueryVariadicOffset(count, offset)
+		for _, v := range excludeEventIDs {
+			params, offset = append(params, v), offset+1
+		}
+	}
 	switch order {
 	case FilterOrderAsc:
 		query += " ORDER BY id ASC"

syncapi/storage/sqlite3/output_room_events_table.go

@@ -150,7 +150,7 @@ func (s *outputRoomEventsStatements) SelectStateInRange(
 		},
 		stateFilter.Senders, stateFilter.NotSenders,
 		stateFilter.Types, stateFilter.NotTypes,
-		stateFilter.Limit, FilterOrderAsc,
+		nil, stateFilter.Limit, FilterOrderAsc,
 	)
 	if err != nil {
 		return nil, nil, fmt.Errorf("s.prepareWithFilters: %w", err)
@@ -326,7 +326,7 @@ func (s *outputRoomEventsStatements) SelectRecentEvents(
 		},
 		eventFilter.Senders, eventFilter.NotSenders,
 		eventFilter.Types, eventFilter.NotTypes,
-		eventFilter.Limit+1, FilterOrderDesc,
+		nil, eventFilter.Limit+1, FilterOrderDesc,
 	)
 	if err != nil {
 		return nil, false, fmt.Errorf("s.prepareWithFilters: %w", err)
@@ -374,7 +374,7 @@ func (s *outputRoomEventsStatements) SelectEarlyEvents(
 		},
 		eventFilter.Senders, eventFilter.NotSenders,
 		eventFilter.Types, eventFilter.NotTypes,
-		eventFilter.Limit, FilterOrderAsc,
+		nil, eventFilter.Limit, FilterOrderAsc,
 	)
 	if err != nil {
 		return nil, fmt.Errorf("s.prepareWithFilters: %w", err)

syncapi/storage/tables/interface.go

@@ -91,7 +91,7 @@ type CurrentRoomState interface {
 	DeleteRoomStateByEventID(ctx context.Context, txn *sql.Tx, eventID string) error
 	DeleteRoomStateForRoom(ctx context.Context, txn *sql.Tx, roomID string) error
 	// SelectCurrentState returns all the current state events for the given room.
-	SelectCurrentState(ctx context.Context, txn *sql.Tx, roomID string, stateFilter *gomatrixserverlib.StateFilter) ([]*gomatrixserverlib.HeaderedEvent, error)
+	SelectCurrentState(ctx context.Context, txn *sql.Tx, roomID string, stateFilter *gomatrixserverlib.StateFilter, excludeEventIDs []string) ([]*gomatrixserverlib.HeaderedEvent, error)
 	// SelectRoomIDsWithMembership returns the list of room IDs which have the given user in the given membership state.
 	SelectRoomIDsWithMembership(ctx context.Context, txn *sql.Tx, userID string, membership string) ([]string, error)
 	// SelectJoinedUsers returns a map of room ID to a list of joined user IDs.

syncapi/streams/stream_pdu.go

@@ -98,7 +98,7 @@ func (p *PDUStreamProvider) CompleteSync(
 
 			var jr *types.JoinResponse
 			jr, err = p.getJoinResponseForCompleteSync(
-				ctx, roomID, r, &stateFilter, &eventFilter, req.Device,
+				ctx, roomID, r, &stateFilter, &eventFilter, req.WantFullState, req.Device,
 			)
 			if err != nil {
 				req.Log.WithError(err).Error("p.getJoinResponseForCompleteSync failed")
@@ -124,7 +124,7 @@ func (p *PDUStreamProvider) CompleteSync(
 		if !peek.Deleted {
 			var jr *types.JoinResponse
 			jr, err = p.getJoinResponseForCompleteSync(
-				ctx, peek.RoomID, r, &stateFilter, &eventFilter, req.Device,
+				ctx, peek.RoomID, r, &stateFilter, &eventFilter, req.WantFullState, req.Device,
 			)
 			if err != nil {
 				req.Log.WithError(err).Error("p.getJoinResponseForCompleteSync failed")
@@ -254,26 +254,37 @@ func (p *PDUStreamProvider) addRoomDeltaToResponse(
 	return nil
 }
 
+// nolint:gocyclo
 func (p *PDUStreamProvider) getJoinResponseForCompleteSync(
 	ctx context.Context,
 	roomID string,
 	r types.Range,
 	stateFilter *gomatrixserverlib.StateFilter,
 	eventFilter *gomatrixserverlib.RoomEventFilter,
+	wantFullState bool,
 	device *userapi.Device,
 ) (jr *types.JoinResponse, err error) {
-	var stateEvents []*gomatrixserverlib.HeaderedEvent
-	stateEvents, err = p.DB.CurrentState(ctx, roomID, stateFilter)
+	// TODO: When filters are added, we may need to call this multiple times to get enough events.
+	//       See: https://github.com/matrix-org/synapse/blob/v0.19.3/synapse/handlers/sync.py#L316
+	recentStreamEvents, limited, err := p.DB.RecentEvents(
+		ctx, roomID, r, eventFilter, true, true,
+	)
 	if err != nil {
 		return
 	}
-	// TODO: When filters are added, we may need to call this multiple times to get enough events.
-	//       See: https://github.com/matrix-org/synapse/blob/v0.19.3/synapse/handlers/sync.py#L316
-	var recentStreamEvents []types.StreamEvent
-	var limited bool
-	recentStreamEvents, limited, err = p.DB.RecentEvents(
-		ctx, roomID, r, eventFilter, true, true,
-	)
+
+	// Get the event IDs of the stream events we fetched. There's no point in us
+	var excludingEventIDs []string
+	if !wantFullState {
+		excludingEventIDs = make([]string, 0, len(recentStreamEvents))
+		for _, event := range recentStreamEvents {
+			if event.StateKey() != nil {
+				excludingEventIDs = append(excludingEventIDs, event.EventID())
+			}
+		}
+	}
+
+	stateEvents, err := p.DB.CurrentState(ctx, roomID, stateFilter, excludingEventIDs)
 	if err != nil {
 		return
 	}