diff --git a/syncapi/storage/postgres/deltas/20201211125500_sequences.go b/syncapi/storage/postgres/deltas/20201211125500_sequences.go new file mode 100644 index 000000000..a51df26f3 --- /dev/null +++ b/syncapi/storage/postgres/deltas/20201211125500_sequences.go @@ -0,0 +1,66 @@ +// Copyright 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 deltas + +import ( + "database/sql" + "fmt" + + "github.com/matrix-org/dendrite/internal/sqlutil" + "github.com/pressly/goose" +) + +func LoadFromGoose() { + goose.AddMigration(UpFixSequences, DownFixSequences) +} + +func LoadFixSequences(m *sqlutil.Migrations) { + m.AddMigration(UpFixSequences, DownFixSequences) +} + +func UpFixSequences(tx *sql.Tx) error { + _, err := tx.Exec(` + -- We need to delete all of the existing receipts because the indexes + -- will be wrong, and we'll get primary key violations if we try to + -- reuse existing stream IDs from a different sequence. + DELETE FROM syncapi_receipts; + + -- Use the new syncapi_receipts_id sequence. + CREATE SEQUENCE IF NOT EXISTS syncapi_receipt_id; + ALTER SEQUENCE IF EXISTS syncapi_receipt_id RESTART WITH 1; + ALTER TABLE syncapi_receipts ALTER COLUMN id SET DEFAULT nextval('syncapi_receipt_id'); + `) + if err != nil { + return fmt.Errorf("failed to execute upgrade: %w", err) + } + return nil +} + +func DownFixSequences(tx *sql.Tx) error { + _, err := tx.Exec(` + -- We need to delete all of the existing receipts because the indexes + -- will be wrong, and we'll get primary key violations if we try to + -- reuse existing stream IDs from a different sequence. + DELETE FROM syncapi_receipts; + + -- Revert back to using the syncapi_stream_id sequence. + DROP SEQUENCE IF EXISTS syncapi_receipt_id; + ALTER TABLE syncapi_receipts ALTER COLUMN id SET DEFAULT nextval('syncapi_stream_id'); + `) + if err != nil { + return fmt.Errorf("failed to execute downgrade: %w", err) + } + return nil +} diff --git a/syncapi/storage/postgres/receipt_table.go b/syncapi/storage/postgres/receipt_table.go index c5ec6cbc6..23c66910f 100644 --- a/syncapi/storage/postgres/receipt_table.go +++ b/syncapi/storage/postgres/receipt_table.go @@ -30,11 +30,12 @@ import ( ) const receiptsSchema = ` -CREATE SEQUENCE IF NOT EXISTS syncapi_stream_id; +CREATE SEQUENCE IF NOT EXISTS syncapi_receipt_id; + -- Stores data about receipts CREATE TABLE IF NOT EXISTS syncapi_receipts ( -- The ID - id BIGINT PRIMARY KEY DEFAULT nextval('syncapi_stream_id'), + id BIGINT PRIMARY KEY DEFAULT nextval('syncapi_receipt_id'), room_id TEXT NOT NULL, receipt_type TEXT NOT NULL, user_id TEXT NOT NULL, @@ -50,7 +51,7 @@ const upsertReceipt = "" + " (room_id, receipt_type, user_id, event_id, receipt_ts)" + " VALUES ($1, $2, $3, $4, $5)" + " ON CONFLICT (room_id, receipt_type, user_id)" + - " DO UPDATE SET id = nextval('syncapi_stream_id'), event_id = $4, receipt_ts = $5" + + " DO UPDATE SET id = nextval('syncapi_receipt_id'), event_id = $4, receipt_ts = $5" + " RETURNING id" const selectRoomReceipts = "" + @@ -58,10 +59,14 @@ const selectRoomReceipts = "" + " FROM syncapi_receipts" + " WHERE room_id = ANY($1) AND id > $2" +const selectMaxReceiptIDSQL = "" + + "SELECT MAX(id) FROM syncapi_receipts" + type receiptStatements struct { db *sql.DB upsertReceipt *sql.Stmt selectRoomReceipts *sql.Stmt + selectMaxReceiptID *sql.Stmt } func NewPostgresReceiptsTable(db *sql.DB) (tables.Receipts, error) { @@ -78,6 +83,9 @@ func NewPostgresReceiptsTable(db *sql.DB) (tables.Receipts, error) { if r.selectRoomReceipts, err = db.Prepare(selectRoomReceipts); err != nil { return nil, fmt.Errorf("unable to prepare selectRoomReceipts statement: %w", err) } + if r.selectMaxReceiptID, err = db.Prepare(selectMaxReceiptIDSQL); err != nil { + return nil, fmt.Errorf("unable to prepare selectRoomReceipts statement: %w", err) + } return r, nil } @@ -104,3 +112,15 @@ func (r *receiptStatements) SelectRoomReceiptsAfter(ctx context.Context, roomIDs } return res, rows.Err() } + +func (s *receiptStatements) SelectMaxReceiptID( + ctx context.Context, txn *sql.Tx, +) (id int64, err error) { + var nullableID sql.NullInt64 + stmt := sqlutil.TxStmt(txn, s.selectMaxReceiptID) + err = stmt.QueryRowContext(ctx).Scan(&nullableID) + if nullableID.Valid { + id = nullableID.Int64 + } + return +} diff --git a/syncapi/storage/postgres/syncserver.go b/syncapi/storage/postgres/syncserver.go index a77c0ec08..60d67ac0e 100644 --- a/syncapi/storage/postgres/syncserver.go +++ b/syncapi/storage/postgres/syncserver.go @@ -23,6 +23,7 @@ import ( "github.com/matrix-org/dendrite/eduserver/cache" "github.com/matrix-org/dendrite/internal/sqlutil" "github.com/matrix-org/dendrite/setup/config" + "github.com/matrix-org/dendrite/syncapi/storage/postgres/deltas" "github.com/matrix-org/dendrite/syncapi/storage/shared" ) @@ -36,6 +37,7 @@ type SyncServerDatasource struct { } // NewDatabase creates a new sync server database +// nolint:gocyclo func NewDatabase(dbProperties *config.DatabaseOptions) (*SyncServerDatasource, error) { var d SyncServerDatasource var err error @@ -86,6 +88,11 @@ func NewDatabase(dbProperties *config.DatabaseOptions) (*SyncServerDatasource, e if err != nil { return nil, err } + m := sqlutil.NewMigrations() + deltas.LoadFixSequences(m) + if err = m.RunDeltas(d.db, dbProperties); err != nil { + return nil, err + } d.Database = shared.Database{ DB: d.db, Writer: d.writer, diff --git a/syncapi/storage/shared/syncserver.go b/syncapi/storage/shared/syncserver.go index c0ae3d7a9..71a42003e 100644 --- a/syncapi/storage/shared/syncserver.go +++ b/syncapi/storage/shared/syncserver.go @@ -483,10 +483,15 @@ func (d *Database) syncPositionTx( if maxPeekID > maxEventID { maxEventID = maxPeekID } + maxReceiptID, err := d.Receipts.SelectMaxReceiptID(ctx, txn) + if err != nil { + return sp, err + } // TODO: complete these positions sp = types.StreamingToken{ - PDUPosition: types.StreamPosition(maxEventID), - TypingPosition: types.StreamPosition(d.EDUCache.GetLatestSyncPosition()), + PDUPosition: types.StreamPosition(maxEventID), + TypingPosition: types.StreamPosition(d.EDUCache.GetLatestSyncPosition()), + ReceiptPosition: types.StreamPosition(maxReceiptID), } return } diff --git a/syncapi/storage/sqlite3/deltas/20201211125500_sequences.go b/syncapi/storage/sqlite3/deltas/20201211125500_sequences.go new file mode 100644 index 000000000..649050135 --- /dev/null +++ b/syncapi/storage/sqlite3/deltas/20201211125500_sequences.go @@ -0,0 +1,58 @@ +// Copyright 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 deltas + +import ( + "database/sql" + "fmt" + + "github.com/matrix-org/dendrite/internal/sqlutil" + "github.com/pressly/goose" +) + +func LoadFromGoose() { + goose.AddMigration(UpFixSequences, DownFixSequences) +} + +func LoadFixSequences(m *sqlutil.Migrations) { + m.AddMigration(UpFixSequences, DownFixSequences) +} + +func UpFixSequences(tx *sql.Tx) error { + _, err := tx.Exec(` + -- We need to delete all of the existing receipts because the indexes + -- will be wrong, and we'll get primary key violations if we try to + -- reuse existing stream IDs from a different sequence. + DELETE FROM syncapi_receipts; + UPDATE syncapi_stream_id SET stream_id=1 WHERE stream_name="receipt"; + `) + if err != nil { + return fmt.Errorf("failed to execute upgrade: %w", err) + } + return nil +} + +func DownFixSequences(tx *sql.Tx) error { + _, err := tx.Exec(` + -- We need to delete all of the existing receipts because the indexes + -- will be wrong, and we'll get primary key violations if we try to + -- reuse existing stream IDs from a different sequence. + DELETE FROM syncapi_receipts; + `) + if err != nil { + return fmt.Errorf("failed to execute downgrade: %w", err) + } + return nil +} diff --git a/syncapi/storage/sqlite3/receipt_table.go b/syncapi/storage/sqlite3/receipt_table.go index b1770e801..dfde1fd2d 100644 --- a/syncapi/storage/sqlite3/receipt_table.go +++ b/syncapi/storage/sqlite3/receipt_table.go @@ -55,11 +55,15 @@ const selectRoomReceipts = "" + " FROM syncapi_receipts" + " WHERE id > $1 and room_id in ($2)" +const selectMaxReceiptIDSQL = "" + + "SELECT MAX(id) FROM syncapi_receipts" + type receiptStatements struct { db *sql.DB streamIDStatements *streamIDStatements upsertReceipt *sql.Stmt selectRoomReceipts *sql.Stmt + selectMaxReceiptID *sql.Stmt } func NewSqliteReceiptsTable(db *sql.DB, streamID *streamIDStatements) (tables.Receipts, error) { @@ -77,12 +81,15 @@ func NewSqliteReceiptsTable(db *sql.DB, streamID *streamIDStatements) (tables.Re if r.selectRoomReceipts, err = db.Prepare(selectRoomReceipts); err != nil { return nil, fmt.Errorf("unable to prepare selectRoomReceipts statement: %w", err) } + if r.selectMaxReceiptID, err = db.Prepare(selectMaxReceiptIDSQL); err != nil { + return nil, fmt.Errorf("unable to prepare selectRoomReceipts statement: %w", err) + } return r, nil } // UpsertReceipt creates new user receipts func (r *receiptStatements) UpsertReceipt(ctx context.Context, txn *sql.Tx, roomId, receiptType, userId, eventId string, timestamp gomatrixserverlib.Timestamp) (pos types.StreamPosition, err error) { - pos, err = r.streamIDStatements.nextStreamID(ctx, txn) + pos, err = r.streamIDStatements.nextReceiptID(ctx, txn) if err != nil { return } @@ -116,3 +123,15 @@ func (r *receiptStatements) SelectRoomReceiptsAfter(ctx context.Context, roomIDs } return res, rows.Err() } + +func (s *receiptStatements) SelectMaxReceiptID( + ctx context.Context, txn *sql.Tx, +) (id int64, err error) { + var nullableID sql.NullInt64 + stmt := sqlutil.TxStmt(txn, s.selectMaxReceiptID) + err = stmt.QueryRowContext(ctx).Scan(&nullableID) + if nullableID.Valid { + id = nullableID.Int64 + } + return +} diff --git a/syncapi/storage/sqlite3/stream_id_table.go b/syncapi/storage/sqlite3/stream_id_table.go index e6bdc4fcb..f73be422d 100644 --- a/syncapi/storage/sqlite3/stream_id_table.go +++ b/syncapi/storage/sqlite3/stream_id_table.go @@ -18,6 +18,8 @@ CREATE TABLE IF NOT EXISTS syncapi_stream_id ( ); INSERT INTO syncapi_stream_id (stream_name, stream_id) VALUES ("global", 0) ON CONFLICT DO NOTHING; +INSERT INTO syncapi_stream_id (stream_name, stream_id) VALUES ("receipt", 0) + ON CONFLICT DO NOTHING; ` const increaseStreamIDStmt = "" + @@ -56,3 +58,13 @@ func (s *streamIDStatements) nextStreamID(ctx context.Context, txn *sql.Tx) (pos err = selectStmt.QueryRowContext(ctx, "global").Scan(&pos) return } + +func (s *streamIDStatements) nextReceiptID(ctx context.Context, txn *sql.Tx) (pos types.StreamPosition, err error) { + increaseStmt := sqlutil.TxStmt(txn, s.increaseStreamIDStmt) + selectStmt := sqlutil.TxStmt(txn, s.selectStreamIDStmt) + if _, err = increaseStmt.ExecContext(ctx, "receipt"); err != nil { + return + } + err = selectStmt.QueryRowContext(ctx, "receipt").Scan(&pos) + return +} diff --git a/syncapi/storage/sqlite3/syncserver.go b/syncapi/storage/sqlite3/syncserver.go index 7df71b384..1ad0e9473 100644 --- a/syncapi/storage/sqlite3/syncserver.go +++ b/syncapi/storage/sqlite3/syncserver.go @@ -25,6 +25,7 @@ import ( "github.com/matrix-org/dendrite/internal/sqlutil" "github.com/matrix-org/dendrite/setup/config" "github.com/matrix-org/dendrite/syncapi/storage/shared" + "github.com/matrix-org/dendrite/syncapi/storage/sqlite3/deltas" ) // SyncServerDatasource represents a sync server datasource which manages @@ -46,13 +47,14 @@ func NewDatabase(dbProperties *config.DatabaseOptions) (*SyncServerDatasource, e return nil, err } d.writer = sqlutil.NewExclusiveWriter() - if err = d.prepare(); err != nil { + if err = d.prepare(dbProperties); err != nil { return nil, err } return &d, nil } -func (d *SyncServerDatasource) prepare() (err error) { +// nolint:gocyclo +func (d *SyncServerDatasource) prepare(dbProperties *config.DatabaseOptions) (err error) { if err = d.PartitionOffsetStatements.Prepare(d.db, d.writer, "syncapi"); err != nil { return err } @@ -99,6 +101,11 @@ func (d *SyncServerDatasource) prepare() (err error) { if err != nil { return err } + m := sqlutil.NewMigrations() + deltas.LoadFixSequences(m) + if err = m.RunDeltas(d.db, dbProperties); err != nil { + return err + } d.Database = shared.Database{ DB: d.db, Writer: d.writer, diff --git a/syncapi/storage/tables/interface.go b/syncapi/storage/tables/interface.go index a2d8791b6..7a166d439 100644 --- a/syncapi/storage/tables/interface.go +++ b/syncapi/storage/tables/interface.go @@ -161,4 +161,5 @@ type Filter interface { type Receipts interface { UpsertReceipt(ctx context.Context, txn *sql.Tx, roomId, receiptType, userId, eventId string, timestamp gomatrixserverlib.Timestamp) (pos types.StreamPosition, err error) SelectRoomReceiptsAfter(ctx context.Context, roomIDs []string, streamPos types.StreamPosition) ([]eduAPI.OutputReceiptEvent, error) + SelectMaxReceiptID(ctx context.Context, txn *sql.Tx) (id int64, err error) }