From a6f995eb45f0d012f79aa66f727cb6260455b334 Mon Sep 17 00:00:00 2001 From: Kegsay Date: Thu, 28 May 2020 11:15:21 +0100 Subject: [PATCH] Merge Updater structs (#1069) * Move Updater structs to shared and use it for postgres * Add constructors for NewXXXUpdater and a useTxns flag In sqlite, we set useTxns=false and comment why. * Handle nil txn * Handle nil in transaction * Missed one * Close the txn at the right time * Don't close the transaction as we reuse it between calls --- go.sum | 2 - roomserver/storage/postgres/storage.go | 390 ++---------------- .../storage/shared/membership_updater.go | 183 ++++++++ roomserver/storage/shared/prepare.go | 24 ++ .../shared/room_recent_events_updater.go | 120 ++++++ roomserver/storage/shared/storage.go | 13 + roomserver/storage/sqlite3/invite_table.go | 5 +- roomserver/storage/sqlite3/storage.go | 388 +---------------- 8 files changed, 381 insertions(+), 744 deletions(-) create mode 100644 roomserver/storage/shared/membership_updater.go create mode 100644 roomserver/storage/shared/room_recent_events_updater.go diff --git a/go.sum b/go.sum index aa0c8dd57..3398195b6 100644 --- a/go.sum +++ b/go.sum @@ -352,8 +352,6 @@ github.com/matrix-org/dugong v0.0.0-20171220115018-ea0a4690a0d5/go.mod h1:NgPCr+ github.com/matrix-org/dugong v0.0.0-20171220115018-ea0a4690a0d5/go.mod h1:NgPCr+UavRGH6n5jmdX8DuqFZ4JiCWIJoZiuhTRLSUg= github.com/matrix-org/go-http-js-libp2p v0.0.0-20200518170932-783164aeeda4 h1:eqE5OnGx9ZMWmrRbD3KF/3KtTunw0iQulI7YxOIdxo4= github.com/matrix-org/go-http-js-libp2p v0.0.0-20200518170932-783164aeeda4/go.mod h1:3WluEZ9QXSwU30tWYqktnpC1x9mwZKx1r8uAv8Iq+a4= -github.com/matrix-org/go-sqlite3-js v0.0.0-20200326102434-98eda28055bd h1:C1FV4dRKF1uuGK8UH01+IoW6zZpfsTV1MvQimZvt418= -github.com/matrix-org/go-sqlite3-js v0.0.0-20200326102434-98eda28055bd/go.mod h1:e+cg2q7C7yE5QnAXgzo512tgFh1RbQLC0+jozuegKgo= github.com/matrix-org/go-sqlite3-js v0.0.0-20200522092705-bc8506ccbcf3 h1:Yb+Wlf/iHhWlLWd+kCgG+Fsg4Dc+xBl7hptfK7lD0zY= github.com/matrix-org/go-sqlite3-js v0.0.0-20200522092705-bc8506ccbcf3/go.mod h1:e+cg2q7C7yE5QnAXgzo512tgFh1RbQLC0+jozuegKgo= github.com/matrix-org/gomatrix v0.0.0-20190528120928-7df988a63f26 h1:Hr3zjRsq2bhrnp3Ky1qgx/fzCtCALOoGYylh2tpS9K4= diff --git a/roomserver/storage/postgres/storage.go b/roomserver/storage/postgres/storage.go index 53a58076a..971f2b9e6 100644 --- a/roomserver/storage/postgres/storage.go +++ b/roomserver/storage/postgres/storage.go @@ -9,14 +9,13 @@ // // 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. +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implie // 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" @@ -25,423 +24,84 @@ import ( // Import the postgres database driver. _ "github.com/lib/pq" "github.com/matrix-org/dendrite/roomserver/storage/shared" - "github.com/matrix-org/dendrite/roomserver/storage/tables" - "github.com/matrix-org/dendrite/roomserver/types" - "github.com/matrix-org/gomatrixserverlib" ) // A Database is used to store room events and stream offsets. type Database struct { shared.Database - events tables.Events - eventTypes tables.EventTypes - eventStateKeys tables.EventStateKeys - eventJSON tables.EventJSON - rooms tables.Rooms - transactions tables.Transactions - prevEvents tables.PreviousEvents - invites tables.Invites - membership tables.Membership - db *sql.DB } // Open a postgres database. // nolint: gocyclo func Open(dataSourceName string, dbProperties internal.DbProperties) (*Database, error) { var d Database + var db *sql.DB var err error - if d.db, err = sqlutil.Open("postgres", dataSourceName, dbProperties); err != nil { + if db, err = sqlutil.Open("postgres", dataSourceName, dbProperties); err != nil { return nil, err } - d.eventStateKeys, err = NewPostgresEventStateKeysTable(d.db) + eventStateKeys, err := NewPostgresEventStateKeysTable(db) if err != nil { return nil, err } - d.eventTypes, err = NewPostgresEventTypesTable(d.db) + eventTypes, err := NewPostgresEventTypesTable(db) if err != nil { return nil, err } - d.eventJSON, err = NewPostgresEventJSONTable(d.db) + eventJSON, err := NewPostgresEventJSONTable(db) if err != nil { return nil, err } - d.events, err = NewPostgresEventsTable(d.db) + events, err := NewPostgresEventsTable(db) if err != nil { return nil, err } - d.rooms, err = NewPostgresRoomsTable(d.db) + rooms, err := NewPostgresRoomsTable(db) if err != nil { return nil, err } - d.transactions, err = NewPostgresTransactionsTable(d.db) + transactions, err := NewPostgresTransactionsTable(db) if err != nil { return nil, err } - stateBlock, err := NewPostgresStateBlockTable(d.db) + stateBlock, err := NewPostgresStateBlockTable(db) if err != nil { return nil, err } - stateSnapshot, err := NewPostgresStateSnapshotTable(d.db) + stateSnapshot, err := NewPostgresStateSnapshotTable(db) if err != nil { return nil, err } - roomAliases, err := NewPostgresRoomAliasesTable(d.db) + roomAliases, err := NewPostgresRoomAliasesTable(db) if err != nil { return nil, err } - d.prevEvents, err = NewPostgresPreviousEventsTable(d.db) + prevEvents, err := NewPostgresPreviousEventsTable(db) if err != nil { return nil, err } - d.invites, err = NewPostgresInvitesTable(d.db) + invites, err := NewPostgresInvitesTable(db) if err != nil { return nil, err } - d.membership, err = NewPostgresMembershipTable(d.db) + membership, err := NewPostgresMembershipTable(db) if err != nil { return nil, err } d.Database = shared.Database{ - DB: d.db, - EventTypesTable: d.eventTypes, - EventStateKeysTable: d.eventStateKeys, - EventJSONTable: d.eventJSON, - EventsTable: d.events, - RoomsTable: d.rooms, - TransactionsTable: d.transactions, + DB: db, + EventTypesTable: eventTypes, + EventStateKeysTable: eventStateKeys, + EventJSONTable: eventJSON, + EventsTable: events, + RoomsTable: rooms, + TransactionsTable: transactions, StateBlockTable: stateBlock, StateSnapshotTable: stateSnapshot, - PrevEventsTable: d.prevEvents, + PrevEventsTable: prevEvents, RoomAliasesTable: roomAliases, - InvitesTable: d.invites, - MembershipTable: d.membership, + InvitesTable: invites, + MembershipTable: membership, } return &d, nil } - -func (d *Database) assignRoomNID( - ctx context.Context, txn *sql.Tx, - roomID string, roomVersion gomatrixserverlib.RoomVersion, -) (types.RoomNID, error) { - // Check if we already have a numeric ID in the database. - roomNID, err := d.rooms.SelectRoomNID(ctx, txn, roomID) - if err == sql.ErrNoRows { - // We don't have a numeric ID so insert one into the database. - roomNID, err = d.rooms.InsertRoomNID(ctx, txn, roomID, roomVersion) - if err == sql.ErrNoRows { - // We raced with another insert so run the select again. - roomNID, err = d.rooms.SelectRoomNID(ctx, txn, roomID) - } - } - return roomNID, err -} - -func (d *Database) assignStateKeyNID( - ctx context.Context, txn *sql.Tx, eventStateKey string, -) (types.EventStateKeyNID, error) { - // Check if we already have a numeric ID in the database. - eventStateKeyNID, err := d.eventStateKeys.SelectEventStateKeyNID(ctx, txn, eventStateKey) - if err == sql.ErrNoRows { - // We don't have a numeric ID so insert one into the database. - eventStateKeyNID, err = d.eventStateKeys.InsertEventStateKeyNID(ctx, txn, eventStateKey) - if err == sql.ErrNoRows { - // We raced with another insert so run the select again. - eventStateKeyNID, err = d.eventStateKeys.SelectEventStateKeyNID(ctx, txn, eventStateKey) - } - } - return eventStateKeyNID, err -} - -// GetLatestEventsForUpdate implements input.EventDatabase -func (d *Database) GetLatestEventsForUpdate( - ctx context.Context, roomNID types.RoomNID, -) (types.RoomRecentEventsUpdater, error) { - txn, err := d.db.Begin() - if err != nil { - return nil, err - } - eventNIDs, lastEventNIDSent, currentStateSnapshotNID, err := - d.rooms.SelectLatestEventsNIDsForUpdate(ctx, txn, roomNID) - if err != nil { - txn.Rollback() // nolint: errcheck - return nil, err - } - stateAndRefs, err := d.events.BulkSelectStateAtEventAndReference(ctx, txn, eventNIDs) - if err != nil { - txn.Rollback() // nolint: errcheck - return nil, err - } - var lastEventIDSent string - if lastEventNIDSent != 0 { - lastEventIDSent, err = d.events.SelectEventID(ctx, txn, lastEventNIDSent) - if err != nil { - txn.Rollback() // nolint: errcheck - return nil, err - } - } - return &roomRecentEventsUpdater{ - transaction{ctx, txn}, d, roomNID, stateAndRefs, lastEventIDSent, currentStateSnapshotNID, - }, nil -} - -type roomRecentEventsUpdater struct { - transaction - d *Database - roomNID types.RoomNID - latestEvents []types.StateAtEventAndReference - lastEventIDSent string - currentStateSnapshotNID types.StateSnapshotNID -} - -// RoomVersion implements types.RoomRecentEventsUpdater -func (u *roomRecentEventsUpdater) RoomVersion() (version gomatrixserverlib.RoomVersion) { - version, _ = u.d.GetRoomVersionForRoomNID(u.ctx, u.roomNID) - return -} - -// LatestEvents implements types.RoomRecentEventsUpdater -func (u *roomRecentEventsUpdater) LatestEvents() []types.StateAtEventAndReference { - return u.latestEvents -} - -// LastEventIDSent implements types.RoomRecentEventsUpdater -func (u *roomRecentEventsUpdater) LastEventIDSent() string { - return u.lastEventIDSent -} - -// CurrentStateSnapshotNID implements types.RoomRecentEventsUpdater -func (u *roomRecentEventsUpdater) CurrentStateSnapshotNID() types.StateSnapshotNID { - return u.currentStateSnapshotNID -} - -// StorePreviousEvents implements types.RoomRecentEventsUpdater -func (u *roomRecentEventsUpdater) StorePreviousEvents(eventNID types.EventNID, previousEventReferences []gomatrixserverlib.EventReference) error { - for _, ref := range previousEventReferences { - if err := u.d.prevEvents.InsertPreviousEvent(u.ctx, u.txn, ref.EventID, ref.EventSHA256, eventNID); err != nil { - return err - } - } - return nil -} - -// IsReferenced implements types.RoomRecentEventsUpdater -func (u *roomRecentEventsUpdater) IsReferenced(eventReference gomatrixserverlib.EventReference) (bool, error) { - err := u.d.prevEvents.SelectPreviousEventExists(u.ctx, u.txn, eventReference.EventID, eventReference.EventSHA256) - if err == nil { - return true, nil - } - if err == sql.ErrNoRows { - return false, nil - } - return false, err -} - -// SetLatestEvents implements types.RoomRecentEventsUpdater -func (u *roomRecentEventsUpdater) SetLatestEvents( - roomNID types.RoomNID, latest []types.StateAtEventAndReference, lastEventNIDSent types.EventNID, - currentStateSnapshotNID types.StateSnapshotNID, -) error { - eventNIDs := make([]types.EventNID, len(latest)) - for i := range latest { - eventNIDs[i] = latest[i].EventNID - } - return u.d.rooms.UpdateLatestEventNIDs(u.ctx, u.txn, roomNID, eventNIDs, lastEventNIDSent, currentStateSnapshotNID) -} - -// HasEventBeenSent implements types.RoomRecentEventsUpdater -func (u *roomRecentEventsUpdater) HasEventBeenSent(eventNID types.EventNID) (bool, error) { - return u.d.events.SelectEventSentToOutput(u.ctx, u.txn, eventNID) -} - -// MarkEventAsSent implements types.RoomRecentEventsUpdater -func (u *roomRecentEventsUpdater) MarkEventAsSent(eventNID types.EventNID) error { - return u.d.events.UpdateEventSentToOutput(u.ctx, u.txn, eventNID) -} - -func (u *roomRecentEventsUpdater) MembershipUpdater(targetUserNID types.EventStateKeyNID, targetLocal bool) (types.MembershipUpdater, error) { - return u.d.membershipUpdaterTxn(u.ctx, u.txn, u.roomNID, targetUserNID, targetLocal) -} - -// MembershipUpdater implements input.RoomEventDatabase -func (d *Database) MembershipUpdater( - ctx context.Context, roomID, targetUserID string, - targetLocal bool, roomVersion gomatrixserverlib.RoomVersion, -) (types.MembershipUpdater, error) { - txn, err := d.db.Begin() - if err != nil { - return nil, err - } - succeeded := false - defer func() { - if !succeeded { - txn.Rollback() // nolint: errcheck - } - }() - - roomNID, err := d.assignRoomNID(ctx, txn, roomID, roomVersion) - if err != nil { - return nil, err - } - - targetUserNID, err := d.assignStateKeyNID(ctx, txn, targetUserID) - if err != nil { - return nil, err - } - - updater, err := d.membershipUpdaterTxn(ctx, txn, roomNID, targetUserNID, targetLocal) - if err != nil { - return nil, err - } - - succeeded = true - return updater, nil -} - -type membershipUpdater struct { - transaction - d *Database - roomNID types.RoomNID - targetUserNID types.EventStateKeyNID - membership tables.MembershipState -} - -func (d *Database) membershipUpdaterTxn( - ctx context.Context, - txn *sql.Tx, - roomNID types.RoomNID, - targetUserNID types.EventStateKeyNID, - targetLocal bool, -) (types.MembershipUpdater, error) { - - if err := d.membership.InsertMembership(ctx, txn, roomNID, targetUserNID, targetLocal); err != nil { - return nil, err - } - - membership, err := d.membership.SelectMembershipForUpdate(ctx, txn, roomNID, targetUserNID) - if err != nil { - return nil, err - } - - return &membershipUpdater{ - transaction{ctx, txn}, d, roomNID, targetUserNID, membership, - }, nil -} - -// IsInvite implements types.MembershipUpdater -func (u *membershipUpdater) IsInvite() bool { - return u.membership == tables.MembershipStateInvite -} - -// IsJoin implements types.MembershipUpdater -func (u *membershipUpdater) IsJoin() bool { - return u.membership == tables.MembershipStateJoin -} - -// IsLeave implements types.MembershipUpdater -func (u *membershipUpdater) IsLeave() bool { - return u.membership == tables.MembershipStateLeaveOrBan -} - -// SetToInvite implements types.MembershipUpdater -func (u *membershipUpdater) SetToInvite(event gomatrixserverlib.Event) (bool, error) { - senderUserNID, err := u.d.assignStateKeyNID(u.ctx, u.txn, event.Sender()) - if err != nil { - return false, err - } - inserted, err := u.d.invites.InsertInviteEvent( - u.ctx, u.txn, event.EventID(), u.roomNID, u.targetUserNID, senderUserNID, event.JSON(), - ) - if err != nil { - return false, err - } - if u.membership != tables.MembershipStateInvite { - if err = u.d.membership.UpdateMembership( - u.ctx, u.txn, u.roomNID, u.targetUserNID, senderUserNID, tables.MembershipStateInvite, 0, - ); err != nil { - return false, err - } - } - return inserted, nil -} - -// SetToJoin implements types.MembershipUpdater -func (u *membershipUpdater) SetToJoin(senderUserID string, eventID string, isUpdate bool) ([]string, error) { - var inviteEventIDs []string - - senderUserNID, err := u.d.assignStateKeyNID(u.ctx, u.txn, senderUserID) - if err != nil { - return nil, err - } - - // If this is a join event update, there is no invite to update - if !isUpdate { - inviteEventIDs, err = u.d.invites.UpdateInviteRetired( - u.ctx, u.txn, u.roomNID, u.targetUserNID, - ) - if err != nil { - return nil, err - } - } - - // Look up the NID of the new join event - nIDs, err := u.d.EventNIDs(u.ctx, []string{eventID}) - if err != nil { - return nil, err - } - - if u.membership != tables.MembershipStateJoin || isUpdate { - if err = u.d.membership.UpdateMembership( - u.ctx, u.txn, u.roomNID, u.targetUserNID, senderUserNID, - tables.MembershipStateJoin, nIDs[eventID], - ); err != nil { - return nil, err - } - } - - return inviteEventIDs, nil -} - -// SetToLeave implements types.MembershipUpdater -func (u *membershipUpdater) SetToLeave(senderUserID string, eventID string) ([]string, error) { - senderUserNID, err := u.d.assignStateKeyNID(u.ctx, u.txn, senderUserID) - if err != nil { - return nil, err - } - inviteEventIDs, err := u.d.invites.UpdateInviteRetired( - u.ctx, u.txn, u.roomNID, u.targetUserNID, - ) - if err != nil { - return nil, err - } - - // Look up the NID of the new leave event - nIDs, err := u.d.EventNIDs(u.ctx, []string{eventID}) - if err != nil { - return nil, err - } - - if u.membership != tables.MembershipStateLeaveOrBan { - if err = u.d.membership.UpdateMembership( - u.ctx, u.txn, u.roomNID, u.targetUserNID, senderUserNID, - tables.MembershipStateLeaveOrBan, nIDs[eventID], - ); err != nil { - return nil, err - } - } - return inviteEventIDs, nil -} - -type transaction struct { - ctx context.Context - txn *sql.Tx -} - -// Commit implements types.Transaction -func (t *transaction) Commit() error { - return t.txn.Commit() -} - -// Rollback implements types.Transaction -func (t *transaction) Rollback() error { - return t.txn.Rollback() -} diff --git a/roomserver/storage/shared/membership_updater.go b/roomserver/storage/shared/membership_updater.go new file mode 100644 index 000000000..5ddf6d84d --- /dev/null +++ b/roomserver/storage/shared/membership_updater.go @@ -0,0 +1,183 @@ +package shared + +import ( + "context" + "database/sql" + + "github.com/matrix-org/dendrite/roomserver/storage/tables" + "github.com/matrix-org/dendrite/roomserver/types" + "github.com/matrix-org/gomatrixserverlib" +) + +type membershipUpdater struct { + transaction + d *Database + roomNID types.RoomNID + targetUserNID types.EventStateKeyNID + membership tables.MembershipState +} + +func NewMembershipUpdater( + ctx context.Context, d *Database, roomID, targetUserID string, + targetLocal bool, roomVersion gomatrixserverlib.RoomVersion, + useTxns bool, +) (types.MembershipUpdater, error) { + txn, err := d.DB.Begin() + if err != nil { + return nil, err + } + succeeded := false + defer func() { + if !succeeded { + txn.Rollback() // nolint: errcheck + } + }() + + roomNID, err := d.assignRoomNID(ctx, txn, roomID, roomVersion) + if err != nil { + return nil, err + } + + targetUserNID, err := d.assignStateKeyNID(ctx, txn, targetUserID) + if err != nil { + return nil, err + } + + updater, err := d.membershipUpdaterTxn(ctx, txn, roomNID, targetUserNID, targetLocal) + if err != nil { + return nil, err + } + + succeeded = true + if !useTxns { + txn.Commit() // nolint: errcheck + updater.transaction.txn = nil + } + return updater, nil +} + +func (d *Database) membershipUpdaterTxn( + ctx context.Context, + txn *sql.Tx, + roomNID types.RoomNID, + targetUserNID types.EventStateKeyNID, + targetLocal bool, +) (*membershipUpdater, error) { + + if err := d.MembershipTable.InsertMembership(ctx, txn, roomNID, targetUserNID, targetLocal); err != nil { + return nil, err + } + + membership, err := d.MembershipTable.SelectMembershipForUpdate(ctx, txn, roomNID, targetUserNID) + if err != nil { + return nil, err + } + + return &membershipUpdater{ + transaction{ctx, txn}, d, roomNID, targetUserNID, membership, + }, nil +} + +// IsInvite implements types.MembershipUpdater +func (u *membershipUpdater) IsInvite() bool { + return u.membership == tables.MembershipStateInvite +} + +// IsJoin implements types.MembershipUpdater +func (u *membershipUpdater) IsJoin() bool { + return u.membership == tables.MembershipStateJoin +} + +// IsLeave implements types.MembershipUpdater +func (u *membershipUpdater) IsLeave() bool { + return u.membership == tables.MembershipStateLeaveOrBan +} + +// SetToInvite implements types.MembershipUpdater +func (u *membershipUpdater) SetToInvite(event gomatrixserverlib.Event) (bool, error) { + senderUserNID, err := u.d.assignStateKeyNID(u.ctx, u.txn, event.Sender()) + if err != nil { + return false, err + } + inserted, err := u.d.InvitesTable.InsertInviteEvent( + u.ctx, u.txn, event.EventID(), u.roomNID, u.targetUserNID, senderUserNID, event.JSON(), + ) + if err != nil { + return false, err + } + if u.membership != tables.MembershipStateInvite { + if err = u.d.MembershipTable.UpdateMembership( + u.ctx, u.txn, u.roomNID, u.targetUserNID, senderUserNID, tables.MembershipStateInvite, 0, + ); err != nil { + return false, err + } + } + return inserted, nil +} + +// SetToJoin implements types.MembershipUpdater +func (u *membershipUpdater) SetToJoin(senderUserID string, eventID string, isUpdate bool) ([]string, error) { + var inviteEventIDs []string + + senderUserNID, err := u.d.assignStateKeyNID(u.ctx, u.txn, senderUserID) + if err != nil { + return nil, err + } + + // If this is a join event update, there is no invite to update + if !isUpdate { + inviteEventIDs, err = u.d.InvitesTable.UpdateInviteRetired( + u.ctx, u.txn, u.roomNID, u.targetUserNID, + ) + if err != nil { + return nil, err + } + } + + // Look up the NID of the new join event + nIDs, err := u.d.EventNIDs(u.ctx, []string{eventID}) + if err != nil { + return nil, err + } + + if u.membership != tables.MembershipStateJoin || isUpdate { + if err = u.d.MembershipTable.UpdateMembership( + u.ctx, u.txn, u.roomNID, u.targetUserNID, senderUserNID, + tables.MembershipStateJoin, nIDs[eventID], + ); err != nil { + return nil, err + } + } + + return inviteEventIDs, nil +} + +// SetToLeave implements types.MembershipUpdater +func (u *membershipUpdater) SetToLeave(senderUserID string, eventID string) ([]string, error) { + senderUserNID, err := u.d.assignStateKeyNID(u.ctx, u.txn, senderUserID) + if err != nil { + return nil, err + } + inviteEventIDs, err := u.d.InvitesTable.UpdateInviteRetired( + u.ctx, u.txn, u.roomNID, u.targetUserNID, + ) + if err != nil { + return nil, err + } + + // Look up the NID of the new leave event + nIDs, err := u.d.EventNIDs(u.ctx, []string{eventID}) + if err != nil { + return nil, err + } + + if u.membership != tables.MembershipStateLeaveOrBan { + if err = u.d.MembershipTable.UpdateMembership( + u.ctx, u.txn, u.roomNID, u.targetUserNID, senderUserNID, + tables.MembershipStateLeaveOrBan, nIDs[eventID], + ); err != nil { + return nil, err + } + } + return inviteEventIDs, nil +} diff --git a/roomserver/storage/shared/prepare.go b/roomserver/storage/shared/prepare.go index 1b3497fd8..65ceec1cc 100644 --- a/roomserver/storage/shared/prepare.go +++ b/roomserver/storage/shared/prepare.go @@ -16,6 +16,7 @@ package shared import ( + "context" "database/sql" ) @@ -34,3 +35,26 @@ func (s StatementList) Prepare(db *sql.DB) (err error) { } return } + +type transaction struct { + ctx context.Context + txn *sql.Tx +} + +// Commit implements types.Transaction +func (t *transaction) Commit() error { + if t.txn == nil { + // The Updater structs can operate in useTxns=false mode. The code will still call this though. + return nil + } + return t.txn.Commit() +} + +// Rollback implements types.Transaction +func (t *transaction) Rollback() error { + if t.txn == nil { + // The Updater structs can operate in useTxns=false mode. The code will still call this though. + return nil + } + return t.txn.Rollback() +} diff --git a/roomserver/storage/shared/room_recent_events_updater.go b/roomserver/storage/shared/room_recent_events_updater.go new file mode 100644 index 000000000..8131f712d --- /dev/null +++ b/roomserver/storage/shared/room_recent_events_updater.go @@ -0,0 +1,120 @@ +package shared + +import ( + "context" + "database/sql" + + "github.com/matrix-org/dendrite/roomserver/types" + "github.com/matrix-org/gomatrixserverlib" +) + +type roomRecentEventsUpdater struct { + transaction + d *Database + roomNID types.RoomNID + latestEvents []types.StateAtEventAndReference + lastEventIDSent string + currentStateSnapshotNID types.StateSnapshotNID +} + +func NewRoomRecentEventsUpdater(d *Database, ctx context.Context, roomNID types.RoomNID, useTxns bool) (types.RoomRecentEventsUpdater, error) { + txn, err := d.DB.Begin() + if err != nil { + return nil, err + } + eventNIDs, lastEventNIDSent, currentStateSnapshotNID, err := + d.RoomsTable.SelectLatestEventsNIDsForUpdate(ctx, txn, roomNID) + if err != nil { + txn.Rollback() // nolint: errcheck + return nil, err + } + stateAndRefs, err := d.EventsTable.BulkSelectStateAtEventAndReference(ctx, txn, eventNIDs) + if err != nil { + txn.Rollback() // nolint: errcheck + return nil, err + } + var lastEventIDSent string + if lastEventNIDSent != 0 { + lastEventIDSent, err = d.EventsTable.SelectEventID(ctx, txn, lastEventNIDSent) + if err != nil { + txn.Rollback() // nolint: errcheck + return nil, err + } + } + if !useTxns { + txn.Commit() // nolint: errcheck + txn = nil + } + return &roomRecentEventsUpdater{ + transaction{ctx, txn}, d, roomNID, stateAndRefs, lastEventIDSent, currentStateSnapshotNID, + }, nil +} + +// RoomVersion implements types.RoomRecentEventsUpdater +func (u *roomRecentEventsUpdater) RoomVersion() (version gomatrixserverlib.RoomVersion) { + version, _ = u.d.GetRoomVersionForRoomNID(u.ctx, u.roomNID) + return +} + +// LatestEvents implements types.RoomRecentEventsUpdater +func (u *roomRecentEventsUpdater) LatestEvents() []types.StateAtEventAndReference { + return u.latestEvents +} + +// LastEventIDSent implements types.RoomRecentEventsUpdater +func (u *roomRecentEventsUpdater) LastEventIDSent() string { + return u.lastEventIDSent +} + +// CurrentStateSnapshotNID implements types.RoomRecentEventsUpdater +func (u *roomRecentEventsUpdater) CurrentStateSnapshotNID() types.StateSnapshotNID { + return u.currentStateSnapshotNID +} + +// StorePreviousEvents implements types.RoomRecentEventsUpdater +func (u *roomRecentEventsUpdater) StorePreviousEvents(eventNID types.EventNID, previousEventReferences []gomatrixserverlib.EventReference) error { + for _, ref := range previousEventReferences { + if err := u.d.PrevEventsTable.InsertPreviousEvent(u.ctx, u.txn, ref.EventID, ref.EventSHA256, eventNID); err != nil { + return err + } + } + return nil +} + +// IsReferenced implements types.RoomRecentEventsUpdater +func (u *roomRecentEventsUpdater) IsReferenced(eventReference gomatrixserverlib.EventReference) (bool, error) { + err := u.d.PrevEventsTable.SelectPreviousEventExists(u.ctx, u.txn, eventReference.EventID, eventReference.EventSHA256) + if err == nil { + return true, nil + } + if err == sql.ErrNoRows { + return false, nil + } + return false, err +} + +// SetLatestEvents implements types.RoomRecentEventsUpdater +func (u *roomRecentEventsUpdater) SetLatestEvents( + roomNID types.RoomNID, latest []types.StateAtEventAndReference, lastEventNIDSent types.EventNID, + currentStateSnapshotNID types.StateSnapshotNID, +) error { + eventNIDs := make([]types.EventNID, len(latest)) + for i := range latest { + eventNIDs[i] = latest[i].EventNID + } + return u.d.RoomsTable.UpdateLatestEventNIDs(u.ctx, u.txn, roomNID, eventNIDs, lastEventNIDSent, currentStateSnapshotNID) +} + +// HasEventBeenSent implements types.RoomRecentEventsUpdater +func (u *roomRecentEventsUpdater) HasEventBeenSent(eventNID types.EventNID) (bool, error) { + return u.d.EventsTable.SelectEventSentToOutput(u.ctx, u.txn, eventNID) +} + +// MarkEventAsSent implements types.RoomRecentEventsUpdater +func (u *roomRecentEventsUpdater) MarkEventAsSent(eventNID types.EventNID) error { + return u.d.EventsTable.UpdateEventSentToOutput(u.ctx, u.txn, eventNID) +} + +func (u *roomRecentEventsUpdater) MembershipUpdater(targetUserNID types.EventStateKeyNID, targetLocal bool) (types.MembershipUpdater, error) { + return u.d.membershipUpdaterTxn(u.ctx, u.txn, u.roomNID, targetUserNID, targetLocal) +} diff --git a/roomserver/storage/shared/storage.go b/roomserver/storage/shared/storage.go index 17bcd6964..bb5b51562 100644 --- a/roomserver/storage/shared/storage.go +++ b/roomserver/storage/shared/storage.go @@ -311,6 +311,19 @@ func (d *Database) GetTransactionEventID( return eventID, err } +func (d *Database) MembershipUpdater( + ctx context.Context, roomID, targetUserID string, + targetLocal bool, roomVersion gomatrixserverlib.RoomVersion, +) (types.MembershipUpdater, error) { + return NewMembershipUpdater(ctx, d, roomID, targetUserID, targetLocal, roomVersion, true) +} + +func (d *Database) GetLatestEventsForUpdate( + ctx context.Context, roomNID types.RoomNID, +) (types.RoomRecentEventsUpdater, error) { + return NewRoomRecentEventsUpdater(d, ctx, roomNID, true) +} + func (d *Database) StoreEvent( ctx context.Context, event gomatrixserverlib.Event, txnAndSessionID *api.TransactionID, authEventNIDs []types.EventNID, diff --git a/roomserver/storage/sqlite3/invite_table.go b/roomserver/storage/sqlite3/invite_table.go index fbe2b0729..7dcc2dc0b 100644 --- a/roomserver/storage/sqlite3/invite_table.go +++ b/roomserver/storage/sqlite3/invite_table.go @@ -90,7 +90,6 @@ func (s *inviteStatements) InsertInviteEvent( inviteEventJSON []byte, ) (bool, error) { stmt := internal.TxStmt(txn, s.insertInviteEventStmt) - defer stmt.Close() // nolint: errcheck result, err := stmt.ExecContext( ctx, inviteEventID, roomNID, targetUserNID, senderUserNID, inviteEventJSON, ) @@ -109,7 +108,7 @@ func (s *inviteStatements) UpdateInviteRetired( txn *sql.Tx, roomNID types.RoomNID, targetUserNID types.EventStateKeyNID, ) (eventIDs []string, err error) { // gather all the event IDs we will retire - stmt := txn.Stmt(s.selectInvitesAboutToRetireStmt) + stmt := internal.TxStmt(txn, s.selectInvitesAboutToRetireStmt) rows, err := stmt.QueryContext(ctx, roomNID, targetUserNID) if err != nil { return nil, err @@ -124,7 +123,7 @@ func (s *inviteStatements) UpdateInviteRetired( } // now retire the invites - stmt = txn.Stmt(s.updateInviteRetiredStmt) + stmt = internal.TxStmt(txn, s.updateInviteRetiredStmt) _, err = stmt.ExecContext(ctx, roomNID, targetUserNID) return } diff --git a/roomserver/storage/sqlite3/storage.go b/roomserver/storage/sqlite3/storage.go index 16d893044..54a2f2658 100644 --- a/roomserver/storage/sqlite3/storage.go +++ b/roomserver/storage/sqlite3/storage.go @@ -140,387 +140,27 @@ func Open(dataSourceName string) (*Database, error) { return &d, nil } -func (d *Database) assignRoomNID( - ctx context.Context, txn *sql.Tx, - roomID string, roomVersion gomatrixserverlib.RoomVersion, -) (roomNID types.RoomNID, err error) { - // Check if we already have a numeric ID in the database. - roomNID, err = d.rooms.SelectRoomNID(ctx, txn, roomID) - if err == sql.ErrNoRows { - // We don't have a numeric ID so insert one into the database. - roomNID, err = d.rooms.InsertRoomNID(ctx, txn, roomID, roomVersion) - if err == nil { - // Now get the numeric ID back out of the database - roomNID, err = d.rooms.SelectRoomNID(ctx, txn, roomID) - } - } - return -} - -func (d *Database) assignStateKeyNID( - ctx context.Context, txn *sql.Tx, eventStateKey string, -) (eventStateKeyNID types.EventStateKeyNID, err error) { - // Check if we already have a numeric ID in the database. - eventStateKeyNID, err = d.eventStateKeys.SelectEventStateKeyNID(ctx, txn, eventStateKey) - if err == sql.ErrNoRows { - // We don't have a numeric ID so insert one into the database. - eventStateKeyNID, err = d.eventStateKeys.InsertEventStateKeyNID(ctx, txn, eventStateKey) - if err == sql.ErrNoRows { - // We raced with another insert so run the select again. - eventStateKeyNID, err = d.eventStateKeys.SelectEventStateKeyNID(ctx, txn, eventStateKey) - } - } - return -} - -// GetLatestEventsForUpdate implements input.EventDatabase func (d *Database) GetLatestEventsForUpdate( ctx context.Context, roomNID types.RoomNID, ) (types.RoomRecentEventsUpdater, error) { - txn, err := d.db.Begin() - if err != nil { - return nil, err - } - eventNIDs, lastEventNIDSent, currentStateSnapshotNID, err := - d.rooms.SelectLatestEventsNIDsForUpdate(ctx, txn, roomNID) - if err != nil { - txn.Rollback() // nolint: errcheck - return nil, err - } - stateAndRefs, err := d.events.BulkSelectStateAtEventAndReference(ctx, txn, eventNIDs) - if err != nil { - txn.Rollback() // nolint: errcheck - return nil, err - } - var lastEventIDSent string - if lastEventNIDSent != 0 { - lastEventIDSent, err = d.events.SelectEventID(ctx, txn, lastEventNIDSent) - if err != nil { - txn.Rollback() // nolint: errcheck - return nil, err - } - } - - // FIXME: we probably want to support long-lived txns in sqlite somehow, but we don't because we get - // 'database is locked' errors caused by multiple write txns (one being the long-lived txn created here) - // so for now let's not use a long-lived txn at all, and just commit it here and set the txn to nil so - // we fail fast if someone tries to use the underlying txn object. - err = txn.Commit() - if err != nil { - return nil, err - } - return &roomRecentEventsUpdater{ - transaction{ctx, nil}, d, roomNID, stateAndRefs, lastEventIDSent, currentStateSnapshotNID, - }, nil + // TODO: Do not use transactions. We should be holding open this transaction but we cannot have + // multiple write transactions on sqlite. The code will perform additional + // write transactions independent of this one which will consistently cause + // 'database is locked' errors. As sqlite doesn't support multi-process on the + // same DB anyway, and we only execute updates sequentially, the only worries + // are for rolling back when things go wrong. (atomicity) + return shared.NewRoomRecentEventsUpdater(&d.Database, ctx, roomNID, false) } -type roomRecentEventsUpdater struct { - transaction - d *Database - roomNID types.RoomNID - latestEvents []types.StateAtEventAndReference - lastEventIDSent string - currentStateSnapshotNID types.StateSnapshotNID -} - -// RoomVersion implements types.RoomRecentEventsUpdater -func (u *roomRecentEventsUpdater) RoomVersion() (version gomatrixserverlib.RoomVersion) { - version, _ = u.d.GetRoomVersionForRoomNID(u.ctx, u.roomNID) - return -} - -// LatestEvents implements types.RoomRecentEventsUpdater -func (u *roomRecentEventsUpdater) LatestEvents() []types.StateAtEventAndReference { - return u.latestEvents -} - -// LastEventIDSent implements types.RoomRecentEventsUpdater -func (u *roomRecentEventsUpdater) LastEventIDSent() string { - return u.lastEventIDSent -} - -// CurrentStateSnapshotNID implements types.RoomRecentEventsUpdater -func (u *roomRecentEventsUpdater) CurrentStateSnapshotNID() types.StateSnapshotNID { - return u.currentStateSnapshotNID -} - -// StorePreviousEvents implements types.RoomRecentEventsUpdater -func (u *roomRecentEventsUpdater) StorePreviousEvents(eventNID types.EventNID, previousEventReferences []gomatrixserverlib.EventReference) error { - err := internal.WithTransaction(u.d.db, func(txn *sql.Tx) error { - for _, ref := range previousEventReferences { - if err := u.d.prevEvents.InsertPreviousEvent(u.ctx, txn, ref.EventID, ref.EventSHA256, eventNID); err != nil { - return err - } - } - return nil - }) - return err -} - -// IsReferenced implements types.RoomRecentEventsUpdater -func (u *roomRecentEventsUpdater) IsReferenced(eventReference gomatrixserverlib.EventReference) (res bool, err error) { - err = internal.WithTransaction(u.d.db, func(txn *sql.Tx) error { - err := u.d.prevEvents.SelectPreviousEventExists(u.ctx, txn, eventReference.EventID, eventReference.EventSHA256) - if err == nil { - res = true - err = nil - } - if err == sql.ErrNoRows { - res = false - err = nil - } - return err - }) - return -} - -// SetLatestEvents implements types.RoomRecentEventsUpdater -func (u *roomRecentEventsUpdater) SetLatestEvents( - roomNID types.RoomNID, latest []types.StateAtEventAndReference, lastEventNIDSent types.EventNID, - currentStateSnapshotNID types.StateSnapshotNID, -) error { - err := internal.WithTransaction(u.d.db, func(txn *sql.Tx) error { - eventNIDs := make([]types.EventNID, len(latest)) - for i := range latest { - eventNIDs[i] = latest[i].EventNID - } - return u.d.rooms.UpdateLatestEventNIDs(u.ctx, txn, roomNID, eventNIDs, lastEventNIDSent, currentStateSnapshotNID) - }) - return err -} - -// HasEventBeenSent implements types.RoomRecentEventsUpdater -func (u *roomRecentEventsUpdater) HasEventBeenSent(eventNID types.EventNID) (res bool, err error) { - err = internal.WithTransaction(u.d.db, func(txn *sql.Tx) error { - res, err = u.d.events.SelectEventSentToOutput(u.ctx, txn, eventNID) - return err - }) - return -} - -// MarkEventAsSent implements types.RoomRecentEventsUpdater -func (u *roomRecentEventsUpdater) MarkEventAsSent(eventNID types.EventNID) error { - err := internal.WithTransaction(u.d.db, func(txn *sql.Tx) error { - return u.d.events.UpdateEventSentToOutput(u.ctx, txn, eventNID) - }) - return err -} - -func (u *roomRecentEventsUpdater) MembershipUpdater(targetUserNID types.EventStateKeyNID, targetLocal bool) (mu types.MembershipUpdater, err error) { - err = internal.WithTransaction(u.d.db, func(txn *sql.Tx) error { - mu, err = u.d.membershipUpdaterTxn(u.ctx, txn, u.roomNID, targetUserNID, targetLocal) - return err - }) - return -} - -// MembershipUpdater implements input.RoomEventDatabase func (d *Database) MembershipUpdater( ctx context.Context, roomID, targetUserID string, targetLocal bool, roomVersion gomatrixserverlib.RoomVersion, ) (updater types.MembershipUpdater, err error) { - var txn *sql.Tx - txn, err = d.db.Begin() - if err != nil { - return nil, err - } - succeeded := false - defer func() { - if !succeeded { - txn.Rollback() // nolint: errcheck - } else { - // TODO: We should be holding open this transaction but we cannot have - // multiple write transactions on sqlite. The code will perform additional - // write transactions independent of this one which will consistently cause - // 'database is locked' errors. For now, we'll break up the transaction and - // hope we don't race too catastrophically. Long term, we should be able to - // thread in txn objects where appropriate (either at the interface level or - // bring matrix business logic into the storage layer). - txerr := txn.Commit() - if err == nil && txerr != nil { - err = txerr - } - } - }() - - roomNID, err := d.assignRoomNID(ctx, txn, roomID, roomVersion) - if err != nil { - return nil, err - } - - targetUserNID, err := d.assignStateKeyNID(ctx, txn, targetUserID) - if err != nil { - return nil, err - } - - updater, err = d.membershipUpdaterTxn(ctx, txn, roomNID, targetUserNID, targetLocal) - if err != nil { - return nil, err - } - - succeeded = true - return updater, nil -} - -type membershipUpdater struct { - transaction - d *Database - roomNID types.RoomNID - targetUserNID types.EventStateKeyNID - membership tables.MembershipState -} - -func (d *Database) membershipUpdaterTxn( - ctx context.Context, - txn *sql.Tx, - roomNID types.RoomNID, - targetUserNID types.EventStateKeyNID, - targetLocal bool, -) (types.MembershipUpdater, error) { - - if err := d.membership.InsertMembership(ctx, txn, roomNID, targetUserNID, targetLocal); err != nil { - return nil, err - } - - membership, err := d.membership.SelectMembershipForUpdate(ctx, txn, roomNID, targetUserNID) - if err != nil { - return nil, err - } - - return &membershipUpdater{ - // purposefully set the txn to nil so if we try to use it we panic and fail fast - transaction{ctx, nil}, d, roomNID, targetUserNID, membership, - }, nil -} - -// IsInvite implements types.MembershipUpdater -func (u *membershipUpdater) IsInvite() bool { - return u.membership == tables.MembershipStateInvite -} - -// IsJoin implements types.MembershipUpdater -func (u *membershipUpdater) IsJoin() bool { - return u.membership == tables.MembershipStateJoin -} - -// IsLeave implements types.MembershipUpdater -func (u *membershipUpdater) IsLeave() bool { - return u.membership == tables.MembershipStateLeaveOrBan -} - -// SetToInvite implements types.MembershipUpdater -func (u *membershipUpdater) SetToInvite(event gomatrixserverlib.Event) (inserted bool, err error) { - err = internal.WithTransaction(u.d.db, func(txn *sql.Tx) error { - senderUserNID, err := u.d.assignStateKeyNID(u.ctx, txn, event.Sender()) - if err != nil { - return err - } - inserted, err = u.d.invites.InsertInviteEvent( - u.ctx, txn, event.EventID(), u.roomNID, u.targetUserNID, senderUserNID, event.JSON(), - ) - if err != nil { - return err - } - if u.membership != tables.MembershipStateInvite { - if err = u.d.membership.UpdateMembership( - u.ctx, txn, u.roomNID, u.targetUserNID, senderUserNID, tables.MembershipStateInvite, 0, - ); err != nil { - return err - } - } - return nil - }) - return -} - -// SetToJoin implements types.MembershipUpdater -func (u *membershipUpdater) SetToJoin(senderUserID string, eventID string, isUpdate bool) (inviteEventIDs []string, err error) { - err = internal.WithTransaction(u.d.db, func(txn *sql.Tx) error { - senderUserNID, err := u.d.assignStateKeyNID(u.ctx, txn, senderUserID) - if err != nil { - return err - } - - // If this is a join event update, there is no invite to update - if !isUpdate { - inviteEventIDs, err = u.d.invites.UpdateInviteRetired( - u.ctx, txn, u.roomNID, u.targetUserNID, - ) - if err != nil { - return err - } - } - - // Look up the NID of the new join event - nIDs, err := u.d.EventNIDs(u.ctx, []string{eventID}) - if err != nil { - return err - } - - if u.membership != tables.MembershipStateJoin || isUpdate { - if err = u.d.membership.UpdateMembership( - u.ctx, txn, u.roomNID, u.targetUserNID, senderUserNID, - tables.MembershipStateJoin, nIDs[eventID], - ); err != nil { - return err - } - } - return nil - }) - - return -} - -// SetToLeave implements types.MembershipUpdater -func (u *membershipUpdater) SetToLeave(senderUserID string, eventID string) (inviteEventIDs []string, err error) { - err = internal.WithTransaction(u.d.db, func(txn *sql.Tx) error { - senderUserNID, err := u.d.assignStateKeyNID(u.ctx, txn, senderUserID) - if err != nil { - return err - } - inviteEventIDs, err = u.d.invites.UpdateInviteRetired( - u.ctx, txn, u.roomNID, u.targetUserNID, - ) - if err != nil { - return err - } - - // Look up the NID of the new leave event - nIDs, err := u.d.EventNIDs(u.ctx, []string{eventID}) - if err != nil { - return err - } - - if u.membership != tables.MembershipStateLeaveOrBan { - if err = u.d.membership.UpdateMembership( - u.ctx, txn, u.roomNID, u.targetUserNID, senderUserNID, - tables.MembershipStateLeaveOrBan, nIDs[eventID], - ); err != nil { - return err - } - } - return nil - }) - return -} - -type transaction struct { - ctx context.Context - txn *sql.Tx -} - -// Commit implements types.Transaction -func (t *transaction) Commit() error { - if t.txn == nil { - return nil - } - return t.txn.Commit() -} - -// Rollback implements types.Transaction -func (t *transaction) Rollback() error { - if t.txn == nil { - return nil - } - return t.txn.Rollback() + // TODO: Do not use transactions. We should be holding open this transaction but we cannot have + // multiple write transactions on sqlite. The code will perform additional + // write transactions independent of this one which will consistently cause + // 'database is locked' errors. As sqlite doesn't support multi-process on the + // same DB anyway, and we only execute updates sequentially, the only worries + // are for rolling back when things go wrong. (atomicity) + return shared.NewMembershipUpdater(ctx, &d.Database, roomID, targetUserID, targetLocal, roomVersion, false) }