From 6d1c6f29e05a11830d45f5a229578e91fc012d4b Mon Sep 17 00:00:00 2001 From: Kegsay Date: Fri, 29 Jan 2021 11:36:26 +0000 Subject: [PATCH 1/6] Add m.room.create to invite stripped state (#1740) MSC1772 needs this because the create event contains info on if the room is a space or not. The create event itself isn't sensitive so other people may find this useful too. --- roomserver/internal/perform/perform_invite.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/roomserver/internal/perform/perform_invite.go b/roomserver/internal/perform/perform_invite.go index 085cb02ed..93a52350c 100644 --- a/roomserver/internal/perform/perform_invite.go +++ b/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, From 62a325ded8d6d4fc72553179da81b509179cc342 Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Fri, 29 Jan 2021 16:32:54 +0000 Subject: [PATCH 2/6] Complete sync performance (#1741) * Parallelise PDU stream fetching for complete sync * Fixes * Fixes * Worker queue * Workers * Don't populate device list changes on complete sync * Don't fast-forward typing notifications either on complete sync * Revert "Don't fast-forward typing notifications either on complete sync" This reverts commit 01471f78431cdd840915111f71bd2b5176e584a8. * Comments --- syncapi/streams/stream_devicelist.go | 2 +- syncapi/streams/stream_pdu.go | 71 +++++++++++++++++++++++----- 2 files changed, 61 insertions(+), 12 deletions(-) diff --git a/syncapi/streams/stream_devicelist.go b/syncapi/streams/stream_devicelist.go index c43d50a49..9ea9d088f 100644 --- a/syncapi/streams/stream_devicelist.go +++ b/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( diff --git a/syncapi/streams/stream_pdu.go b/syncapi/streams/stream_pdu.go index d6d7ff444..ae38dc30e 100644 --- a/syncapi/streams/stream_pdu.go +++ b/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 jr *types.JoinResponse - jr, err = p.getJoinResponseForCompleteSync( - ctx, roomID, r, &stateFilter, &eventFilter, req.Device, - ) - if err != nil { - req.Log.WithError(err).Error("p.getJoinResponseForCompleteSync failed") - return from - } - req.Response.Rooms.Join[roomID] = *jr - req.Rooms[roomID] = gomatrixserverlib.Join + var reqMutex sync.Mutex + var reqWaitGroup sync.WaitGroup + reqWaitGroup.Add(len(joinedRoomIDs)) + for _, room := range joinedRoomIDs { + roomID := room + p.queue(func() { + defer reqWaitGroup.Done() + + var jr *types.JoinResponse + jr, err = p.getJoinResponseForCompleteSync( + ctx, roomID, r, &stateFilter, &eventFilter, 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 { From 369d3939fdf88546dfdaf8c185125c684c55e991 Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Fri, 29 Jan 2021 16:33:59 +0000 Subject: [PATCH 3/6] Drop state events we can't auth instead of failing altogether --- go.mod | 2 +- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index 48d0b00d7..96aa881c6 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 999f13c1f..a6464cc8e 100644 --- a/go.sum +++ b/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-20210128131744-c803b6ee2b68/go.mod h1:JsAzE1Ll3+gDWS9JSUHPJiiyAksvOOnGWF2nXdg4ZzU= +github.com/matrix-org/gomatrixserverlib v0.0.0-20210129163316-dd4d53729ead h1:VmGJybKUQin8+NyA9ZkrHJpE8ygXzcON9peQH9LC92c= +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= From de5f22a46960308b3fe8efe7ca1813460b2c9c09 Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Thu, 4 Feb 2021 11:12:52 +0000 Subject: [PATCH 4/6] Remove redundant check (#1748) --- roomserver/internal/input/input_membership.go | 7 ------- 1 file changed, 7 deletions(-) diff --git a/roomserver/internal/input/input_membership.go b/roomserver/internal/input/input_membership.go index 692d8147a..bc646c3c6 100644 --- a/roomserver/internal/input/input_membership.go +++ b/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 From b7e3b81a22775697ad8ff463247d83077ea6cbe3 Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Thu, 4 Feb 2021 11:45:49 +0000 Subject: [PATCH 5/6] Fix ON CONFLICT on sync API account data (#1745) (#1750) --- syncapi/storage/postgres/account_data_table.go | 2 +- syncapi/storage/sqlite3/account_data_table.go | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/syncapi/storage/postgres/account_data_table.go b/syncapi/storage/postgres/account_data_table.go index 67eb1e863..25bdb1da3 100644 --- a/syncapi/storage/postgres/account_data_table.go +++ b/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 = "" + diff --git a/syncapi/storage/sqlite3/account_data_table.go b/syncapi/storage/sqlite3/account_data_table.go index 1c65cb6a9..24c442240 100644 --- a/syncapi/storage/sqlite3/account_data_table.go +++ b/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 } From 6099379ea48cf47f9570e9bc51aba4bb3fa8c066 Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Thu, 4 Feb 2021 11:52:49 +0000 Subject: [PATCH 6/6] Remove rooms table from federation sender (#1751) * Remove last sent event ID column from federation sender * Remove EventIDMismatchError * Remove the federationsender rooms table altogether, it's useless * Add migration * Fix migrations * Fix migrations --- .../postgres/deltas/2021020411080000_rooms.go | 46 ++++++++ .../storage/postgres/room_table.go | 104 ----------------- federationsender/storage/postgres/storage.go | 11 +- federationsender/storage/shared/storage.go | 26 +---- .../sqlite3/deltas/2021020411080000_rooms.go | 46 ++++++++ .../storage/sqlite3/room_table.go | 105 ------------------ federationsender/storage/sqlite3/storage.go | 11 +- federationsender/storage/tables/interface.go | 6 - federationsender/types/types.go | 18 --- 9 files changed, 105 insertions(+), 268 deletions(-) create mode 100644 federationsender/storage/postgres/deltas/2021020411080000_rooms.go delete mode 100644 federationsender/storage/postgres/room_table.go create mode 100644 federationsender/storage/sqlite3/deltas/2021020411080000_rooms.go delete mode 100644 federationsender/storage/sqlite3/room_table.go diff --git a/federationsender/storage/postgres/deltas/2021020411080000_rooms.go b/federationsender/storage/postgres/deltas/2021020411080000_rooms.go new file mode 100644 index 000000000..cc4bdadfd --- /dev/null +++ b/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 +} diff --git a/federationsender/storage/postgres/room_table.go b/federationsender/storage/postgres/room_table.go deleted file mode 100644 index 8d3ed20ff..000000000 --- a/federationsender/storage/postgres/room_table.go +++ /dev/null @@ -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 -} diff --git a/federationsender/storage/postgres/storage.go b/federationsender/storage/postgres/storage.go index b9827ca19..5edc08ad7 100644 --- a/federationsender/storage/postgres/storage.go +++ b/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, diff --git a/federationsender/storage/shared/storage.go b/federationsender/storage/shared/storage.go index 4c9490424..2e74e9d6a 100644 --- a/federationsender/storage/shared/storage.go +++ b/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 } diff --git a/federationsender/storage/sqlite3/deltas/2021020411080000_rooms.go b/federationsender/storage/sqlite3/deltas/2021020411080000_rooms.go new file mode 100644 index 000000000..cc4bdadfd --- /dev/null +++ b/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 +} diff --git a/federationsender/storage/sqlite3/room_table.go b/federationsender/storage/sqlite3/room_table.go deleted file mode 100644 index 0710ccca3..000000000 --- a/federationsender/storage/sqlite3/room_table.go +++ /dev/null @@ -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 -} diff --git a/federationsender/storage/sqlite3/storage.go b/federationsender/storage/sqlite3/storage.go index 2b1358587..84a9ff860 100644 --- a/federationsender/storage/sqlite3/storage.go +++ b/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, diff --git a/federationsender/storage/tables/interface.go b/federationsender/storage/tables/interface.go index 22fd5554f..34ff0b97e 100644 --- a/federationsender/storage/tables/interface.go +++ b/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) diff --git a/federationsender/types/types.go b/federationsender/types/types.go index 90da310c9..c486c05c4 100644 --- a/federationsender/types/types.go +++ b/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