From 55c7f2c892e635740bc99a8cd2318847cc028989 Mon Sep 17 00:00:00 2001 From: Matthew Hodgson Date: Tue, 8 Sep 2020 00:56:50 +0100 Subject: [PATCH] add (broken) postgres; advance streampos whenever sync output changes --- syncapi/storage/postgres/peeks_table.go | 195 ++++++++++++++++++++++++ syncapi/storage/postgres/syncserver.go | 5 + syncapi/storage/shared/syncserver.go | 16 +- syncapi/storage/sqlite3/peeks_table.go | 48 ++++-- syncapi/storage/tables/interface.go | 3 +- 5 files changed, 252 insertions(+), 15 deletions(-) create mode 100644 syncapi/storage/postgres/peeks_table.go diff --git a/syncapi/storage/postgres/peeks_table.go b/syncapi/storage/postgres/peeks_table.go new file mode 100644 index 00000000..c8938a2a --- /dev/null +++ b/syncapi/storage/postgres/peeks_table.go @@ -0,0 +1,195 @@ +// 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 postgres + +import ( + "context" + "database/sql" + "time" + + "github.com/matrix-org/dendrite/internal" + "github.com/matrix-org/dendrite/internal/sqlutil" + "github.com/matrix-org/dendrite/syncapi/storage/tables" + "github.com/matrix-org/dendrite/syncapi/types" +) + +const peeksSchema = ` +CREATE TABLE IF NOT EXISTS syncapi_peeks ( + id BIGINT PRIMARY KEY DEFAULT nextval('syncapi_stream_id'), + room_id TEXT NOT NULL, + user_id TEXT NOT NULL, + device_id TEXT NOT NULL, + new BOOL NOT NULL DEFAULT true, + deleted BOOL NOT NULL DEFAULT false, + -- When the peek was created in UNIX epoch ms. + creation_ts BIGINT NOT NULL +); + +CREATE INDEX IF NOT EXISTS syncapi_peeks_room_id_idx ON syncapi_peeks(room_id); +CREATE INDEX IF NOT EXISTS syncapi_peeks_user_id_device_id_idx ON syncapi_peeks(user_id, device_id); +` + +const insertPeekSQL = "" + + "INSERT INTO syncapi_peeks" + + " (room_id, user_id, device_id, creation_ts)" + + " VALUES ($1, $2, $3, $4) RETURNING id" + +const deletePeekSQL = "" + + "UPDATE syncapi_peeks SET deleted=true, id=nextval('syncapi_stream_id') WHERE room_id = $1 AND user_id = $2 AND device_id = $3 RETURNING id" + +const deletePeeksSQL = "" + + "UPDATE syncapi_peeks SET deleted=true, id=nextval('syncapi_stream_id') WHERE room_id = $1 AND user_id = $2 RETURNING id" + +const selectPeeksSQL = "" + + "SELECT room_id, new FROM syncapi_peeks WHERE user_id = $1 AND device_id = $2 AND deleted=false" + +const selectPeekingDevicesSQL = "" + + "SELECT room_id, user_id, device_id FROM syncapi_peeks WHERE deleted=false" + +const markPeeksAsOldSQL = "" + + "UPDATE syncapi_peeks SET id=nextval('syncapi_stream_id'), new=false WHERE user_id = $1 AND device_id = $2 AND deleted=false RETURNING id" + +const selectMaxPeekIDSQL = "" + + "SELECT MAX(id) FROM syncapi_peeks" + +type peekStatements struct { + db *sql.DB + insertPeekStmt *sql.Stmt + deletePeekStmt *sql.Stmt + deletePeeksStmt *sql.Stmt + selectPeeksStmt *sql.Stmt + selectPeekingDevicesStmt *sql.Stmt + markPeeksAsOldStmt *sql.Stmt + selectMaxPeekIDStmt *sql.Stmt +} + +func NewPostgresPeeksTable(db *sql.DB) (tables.Peeks, error) { + _, err := db.Exec(peeksSchema) + if err != nil { + return nil, err + } + s := &peekStatements{ + db: db, + } + if s.insertPeekStmt, err = db.Prepare(insertPeekSQL); err != nil { + return nil, err + } + if s.deletePeekStmt, err = db.Prepare(deletePeekSQL); err != nil { + return nil, err + } + if s.deletePeeksStmt, err = db.Prepare(deletePeeksSQL); err != nil { + return nil, err + } + if s.selectPeeksStmt, err = db.Prepare(selectPeeksSQL); err != nil { + return nil, err + } + if s.selectPeekingDevicesStmt, err = db.Prepare(selectPeekingDevicesSQL); err != nil { + return nil, err + } + if s.markPeeksAsOldStmt, err = db.Prepare(markPeeksAsOldSQL); err != nil { + return nil, err + } + if s.selectMaxPeekIDStmt, err = db.Prepare(selectMaxPeekIDSQL); err != nil { + return nil, err + } + return s, nil +} + +func (s *peekStatements) InsertPeek( + ctx context.Context, txn *sql.Tx, roomID, userID, deviceID string, +) (streamPos types.StreamPosition, err error) { + nowMilli := time.Now().UnixNano() / int64(time.Millisecond) + stmt := sqlutil.TxStmt(txn, s.insertPeekStmt) + err = stmt.QueryRowContext(ctx, roomID, userID, deviceID, nowMilli).Scan(&streamPos) + return +} + +func (s *peekStatements) DeletePeek( + ctx context.Context, txn *sql.Tx, roomID, userID, deviceID string, +) (streamPos types.StreamPosition, err error) { + stmt := sqlutil.TxStmt(txn, s.deletePeekStmt) + err = stmt.QueryRowContext(ctx, roomID, userID, deviceID).Scan(&streamPos) + return +} + +func (s *peekStatements) DeletePeeks( + ctx context.Context, txn *sql.Tx, roomID, userID string, +) (streamPos types.StreamPosition, err error) { + stmt := sqlutil.TxStmt(txn, s.deletePeeksStmt) + err = stmt.QueryRowContext(ctx, roomID, userID).Scan(&streamPos) + return +} + +func (s *peekStatements) SelectPeeks( + ctx context.Context, txn *sql.Tx, userID, deviceID string, +) (peeks []types.Peek, err error) { + rows, err := sqlutil.TxStmt(txn, s.selectPeeksStmt).QueryContext(ctx, userID, deviceID) + if err != nil { + return + } + defer internal.CloseAndLogIfError(ctx, rows, "SelectPeeks: rows.close() failed") + + for rows.Next() { + peek := types.Peek{} + if err = rows.Scan(&peek.RoomID, &peek.New); err != nil { + return + } + peeks = append(peeks, peek) + } + + return peeks, rows.Err() +} + +func (s *peekStatements) MarkPeeksAsOld( + ctx context.Context, txn *sql.Tx, userID, deviceID string, +) (streamPos types.StreamPosition, err error) { + stmt := sqlutil.TxStmt(txn, s.markPeeksAsOldStmt) + err = stmt.QueryRowContext(ctx, userID, deviceID).Scan(&streamPos) + return +} + +func (s *peekStatements) SelectPeekingDevices( + ctx context.Context, +) (peekingDevices map[string][]types.PeekingDevice, err error) { + rows, err := s.selectPeekingDevicesStmt.QueryContext(ctx) + if err != nil { + return nil, err + } + defer internal.CloseAndLogIfError(ctx, rows, "SelectPeekingDevices: rows.close() failed") + + result := make(map[string][]types.PeekingDevice) + for rows.Next() { + var roomID, userID, deviceID string + if err := rows.Scan(&roomID, &userID, &deviceID); err != nil { + return nil, err + } + devices := result[roomID] + devices = append(devices, types.PeekingDevice{UserID: userID, DeviceID: deviceID}) + result[roomID] = devices + } + return result, nil +} + +func (s *peekStatements) SelectMaxPeekID( + ctx context.Context, txn *sql.Tx, +) (id int64, err error) { + var nullableID sql.NullInt64 + stmt := sqlutil.TxStmt(txn, s.selectMaxPeekIDStmt) + 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 e7f2c944..ad929a40 100644 --- a/syncapi/storage/postgres/syncserver.go +++ b/syncapi/storage/postgres/syncserver.go @@ -62,6 +62,10 @@ func NewDatabase(dbProperties *config.DatabaseOptions) (*SyncServerDatasource, e if err != nil { return nil, err } + peeks, err := NewPostgresPeeksTable(d.db) + if err != nil { + return nil, err + } topology, err := NewPostgresTopologyTable(d.db) if err != nil { return nil, err @@ -82,6 +86,7 @@ func NewDatabase(dbProperties *config.DatabaseOptions) (*SyncServerDatasource, e DB: d.db, Writer: d.writer, Invites: invites, + Peeks: peeks, AccountData: accountData, OutputEvents: events, Topology: topology, diff --git a/syncapi/storage/shared/syncserver.go b/syncapi/storage/shared/syncserver.go index 8170c35c..c8c92753 100644 --- a/syncapi/storage/shared/syncserver.go +++ b/syncapi/storage/shared/syncserver.go @@ -408,6 +408,9 @@ func (d *Database) syncPositionTx( ctx context.Context, txn *sql.Tx, ) (sp types.StreamingToken, err error) { + // XXX: it seems very inefficient to be doing all these aggregate selects + // every time we call /sync... + maxEventID, err := d.OutputEvents.SelectMaxEventID(ctx, txn) if err != nil { return sp, err @@ -426,6 +429,13 @@ func (d *Database) syncPositionTx( if maxInviteID > maxEventID { maxEventID = maxInviteID } + maxPeekID, err := d.Peeks.SelectMaxPeekID(ctx, txn) + if err != nil { + return sp, err + } + if maxPeekID > maxEventID { + maxEventID = maxPeekID + } sp = types.NewStreamToken(types.StreamPosition(maxEventID), types.StreamPosition(d.EDUCache.GetLatestSyncPosition()), nil) return } @@ -1045,7 +1055,8 @@ func (d *Database) getStateDeltas( if newPeeks { err = d.Writer.Do(d.DB, txn, func(txn *sql.Tx) error { - return d.Peeks.MarkPeeksAsOld(ctx, txn, userID, device.ID) + _, err := d.Peeks.MarkPeeksAsOld(ctx, txn, userID, device.ID) + return err }) if err != nil { return nil, nil, err @@ -1150,7 +1161,8 @@ func (d *Database) getStateDeltasForFullStateSync( if newPeeks { err = d.Writer.Do(d.DB, txn, func(txn *sql.Tx) error { - return d.Peeks.MarkPeeksAsOld(ctx, txn, userID, device.ID) + _, err := d.Peeks.MarkPeeksAsOld(ctx, txn, userID, device.ID) + return err }) if err != nil { return nil, nil, err diff --git a/syncapi/storage/sqlite3/peeks_table.go b/syncapi/storage/sqlite3/peeks_table.go index 431d5ed7..c6722901 100644 --- a/syncapi/storage/sqlite3/peeks_table.go +++ b/syncapi/storage/sqlite3/peeks_table.go @@ -32,6 +32,7 @@ CREATE TABLE IF NOT EXISTS syncapi_peeks ( user_id TEXT NOT NULL, device_id TEXT NOT NULL, new BOOL NOT NULL DEFAULT true, + deleted BOOL NOT NULL DEFAULT false, -- When the peek was created in UNIX epoch ms. creation_ts INTEGER NOT NULL ); @@ -42,23 +43,26 @@ CREATE INDEX IF NOT EXISTS syncapi_peeks_user_id_device_id_idx ON syncapi_peeks( const insertPeekSQL = "" + "INSERT INTO syncapi_peeks" + - " (room_id, user_id, device_id, creation_ts)" + - " VALUES ($1, $2, $3, $4)" + " (id, room_id, user_id, device_id, creation_ts)" + + " VALUES ($1, $2, $3, $4, $5)" const deletePeekSQL = "" + - "DELETE FROM syncapi_peeks WHERE room_id = $1 AND user_id = $2 and device_id = $3" + "UPDATE syncapi_peeks SET deleted=true, id=$1 WHERE room_id = $2 AND user_id = $3 AND device_id = $4" const deletePeeksSQL = "" + - "DELETE FROM syncapi_peeks WHERE room_id = $1 AND user_id = $2" + "UPDATE syncapi_peeks SET deleted=true, id=$1 WHERE room_id = $2 AND user_id = $3" const selectPeeksSQL = "" + - "SELECT room_id, new FROM syncapi_peeks WHERE user_id = $1 and device_id = $2" + "SELECT room_id, new FROM syncapi_peeks WHERE user_id = $1 AND device_id = $2 AND deleted=false" const selectPeekingDevicesSQL = "" + - "SELECT room_id, user_id, device_id FROM syncapi_peeks" + "SELECT room_id, user_id, device_id FROM syncapi_peeks WHERE deleted=false" const markPeeksAsOldSQL = "" + - "UPDATE syncapi_peeks SET new=false WHERE user_id = $1 and device_id = $2" + "UPDATE syncapi_peeks SET new=false, id=$1 WHERE user_id = $2 AND device_id = $3 AND deleted=false" + +const selectMaxPeekIDSQL = "" + + "SELECT MAX(id) FROM syncapi_peeks" type peekStatements struct { db *sql.DB @@ -69,6 +73,7 @@ type peekStatements struct { selectPeeksStmt *sql.Stmt selectPeekingDevicesStmt *sql.Stmt markPeeksAsOldStmt *sql.Stmt + selectMaxPeekIDStmt *sql.Stmt } func NewSqlitePeeksTable(db *sql.DB, streamID *streamIDStatements) (tables.Peeks, error) { @@ -98,6 +103,9 @@ func NewSqlitePeeksTable(db *sql.DB, streamID *streamIDStatements) (tables.Peeks if s.markPeeksAsOldStmt, err = db.Prepare(markPeeksAsOldSQL); err != nil { return nil, err } + if s.selectMaxPeekIDStmt, err = db.Prepare(selectMaxPeekIDSQL); err != nil { + return nil, err + } return s, nil } @@ -109,7 +117,7 @@ func (s *peekStatements) InsertPeek( return } nowMilli := time.Now().UnixNano() / int64(time.Millisecond) - _, err = sqlutil.TxStmt(txn, s.insertPeekStmt).ExecContext(ctx, roomID, userID, deviceID, nowMilli) + _, err = sqlutil.TxStmt(txn, s.insertPeekStmt).ExecContext(ctx, streamPos, roomID, userID, deviceID, nowMilli) return } @@ -120,7 +128,7 @@ func (s *peekStatements) DeletePeek( if err != nil { return } - _, err = sqlutil.TxStmt(txn, s.deletePeekStmt).ExecContext(ctx, roomID, userID, deviceID) + _, err = sqlutil.TxStmt(txn, s.deletePeekStmt).ExecContext(ctx, streamPos, roomID, userID, deviceID) return } @@ -131,7 +139,7 @@ func (s *peekStatements) DeletePeeks( if err != nil { return } - _, err = sqlutil.TxStmt(txn, s.deletePeeksStmt).ExecContext(ctx, roomID, userID) + _, err = sqlutil.TxStmt(txn, s.deletePeeksStmt).ExecContext(ctx, streamPos, roomID, userID) return } @@ -157,8 +165,12 @@ func (s *peekStatements) SelectPeeks( func (s *peekStatements) MarkPeeksAsOld( ctx context.Context, txn *sql.Tx, userID, deviceID string, -) (err error) { - _, err = sqlutil.TxStmt(txn, s.markPeeksAsOldStmt).ExecContext(ctx, userID, deviceID) +) (streamPos types.StreamPosition, err error) { + streamPos, err = s.streamIDStatements.nextStreamID(ctx, txn) + if err != nil { + return + } + _, err = sqlutil.TxStmt(txn, s.markPeeksAsOldStmt).ExecContext(ctx, streamPos, userID, deviceID) return } @@ -183,3 +195,15 @@ func (s *peekStatements) SelectPeekingDevices( } return result, nil } + +func (s *peekStatements) SelectMaxPeekID( + ctx context.Context, txn *sql.Tx, +) (id int64, err error) { + var nullableID sql.NullInt64 + stmt := sqlutil.TxStmt(txn, s.selectMaxPeekIDStmt) + err = stmt.QueryRowContext(ctx).Scan(&nullableID) + if nullableID.Valid { + id = nullableID.Int64 + } + return +} diff --git a/syncapi/storage/tables/interface.go b/syncapi/storage/tables/interface.go index 1f8663b9..d189d7e3 100644 --- a/syncapi/storage/tables/interface.go +++ b/syncapi/storage/tables/interface.go @@ -45,7 +45,8 @@ type Peeks interface { DeletePeeks(ctx context.Context, txn *sql.Tx, roomID, userID string) (streamPos types.StreamPosition, err error) SelectPeeks(ctxt context.Context, txn *sql.Tx, userID, deviceID string) (peeks []types.Peek, err error) SelectPeekingDevices(ctxt context.Context) (peekingDevices map[string][]types.PeekingDevice, err error) - MarkPeeksAsOld(ctxt context.Context, txn *sql.Tx, userID, deviceID string) (err error) + MarkPeeksAsOld(ctxt context.Context, txn *sql.Tx, userID, deviceID string) (streamPos types.StreamPosition, err error) + SelectMaxPeekID(ctx context.Context, txn *sql.Tx) (id int64, err error) } type Events interface {