mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2025-02-15 07:24:27 -06:00
Code Issues Packages Projects Releases Wiki Activity

More tweaks

Browse source
This commit is contained in:
Neil Alexander 2022-02-01 14:54:28 +00:00
parent 418339011c
commit fdbb2406be
No known key found for this signature in database
GPG key ID: A02A2019A2BB0944
2 changed files with 17 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
roomserver/storage/shared/room_updater.go
View file

@ -83,19 +83,21 @@ func (u *RoomUpdater) CurrentStateSnapshotNID() types.StateSnapshotNID {
// StorePreviousEvents implements types.RoomRecentEventsUpdater - This must be called from a Writer // StorePreviousEvents implements types.RoomRecentEventsUpdater - This must be called from a Writer
func (u *RoomUpdater) StorePreviousEvents(eventNID types.EventNID, previousEventReferences []gomatrixserverlib.EventReference) error { func (u *RoomUpdater) StorePreviousEvents(eventNID types.EventNID, previousEventReferences []gomatrixserverlib.EventReference) error {
for _, ref := range previousEventReferences { return u.d.Writer.Do(u.d.DB, u.txn, func(txn *sql.Tx) error {
if err := u.d.PrevEventsTable.InsertPreviousEvent(u.ctx, u.txn, ref.EventID, ref.EventSHA256, eventNID); err != nil { for _, ref := range previousEventReferences {
return fmt.Errorf("u.d.PrevEventsTable.InsertPreviousEvent: %w", err) if err := u.d.PrevEventsTable.InsertPreviousEvent(u.ctx, txn, ref.EventID, ref.EventSHA256, eventNID); err != nil {
return fmt.Errorf("u.d.PrevEventsTable.InsertPreviousEvent: %w", err)
}
} }
} return nil
return nil })
} }
func (u *RoomUpdater) StoreEvent( func (u *RoomUpdater) StoreEvent(
ctx context.Context, event *gomatrixserverlib.Event, ctx context.Context, event *gomatrixserverlib.Event,
authEventNIDs []types.EventNID, isRejected bool, authEventNIDs []types.EventNID, isRejected bool,
) (types.EventNID, types.RoomNID, types.StateAtEvent, *gomatrixserverlib.Event, string, error) { ) (types.EventNID, types.RoomNID, types.StateAtEvent, *gomatrixserverlib.Event, string, error) {
return u.d.storeEvent(ctx, u.txn, u, event, authEventNIDs, isRejected) return u.d.storeEvent(ctx, u, event, authEventNIDs, isRejected)
} }
func (u *RoomUpdater) AddState( func (u *RoomUpdater) AddState(

27
roomserver/storage/shared/storage.go
View file

@ -497,11 +497,11 @@ func (d *Database) StoreEvent(
ctx context.Context, event *gomatrixserverlib.Event, ctx context.Context, event *gomatrixserverlib.Event,
authEventNIDs []types.EventNID, isRejected bool, authEventNIDs []types.EventNID, isRejected bool,
) (types.EventNID, types.RoomNID, types.StateAtEvent, *gomatrixserverlib.Event, string, error) { ) (types.EventNID, types.RoomNID, types.StateAtEvent, *gomatrixserverlib.Event, string, error) {
return d.storeEvent(ctx, nil, nil, event, authEventNIDs, isRejected) return d.storeEvent(ctx, nil, event, authEventNIDs, isRejected)
} }
func (d *Database) storeEvent( func (d *Database) storeEvent(
ctx context.Context, txn *sql.Tx, updater *RoomUpdater, event *gomatrixserverlib.Event, ctx context.Context, updater *RoomUpdater, event *gomatrixserverlib.Event,
authEventNIDs []types.EventNID, isRejected bool, authEventNIDs []types.EventNID, isRejected bool,
) (types.EventNID, types.RoomNID, types.StateAtEvent, *gomatrixserverlib.Event, string, error) { ) (types.EventNID, types.RoomNID, types.StateAtEvent, *gomatrixserverlib.Event, string, error) {
var ( var (
@ -514,7 +514,10 @@ func (d *Database) storeEvent(
redactedEventID string redactedEventID string
err error err error
) )
var txn *sql.Tx
if updater != nil {
txn = updater.txn
}
err = d.Writer.Do(d.DB, txn, func(txn *sql.Tx) error { err = d.Writer.Do(d.DB, txn, func(txn *sql.Tx) error {
// TODO: Here we should aim to have two different code paths for new rooms // TODO: Here we should aim to have two different code paths for new rooms
// vs existing ones. // vs existing ones.
@ -610,22 +613,10 @@ func (d *Database) storeEvent(
} }
defer sqlutil.EndTransactionWithCheck(updater.txn, &succeeded, &err) defer sqlutil.EndTransactionWithCheck(updater.txn, &succeeded, &err)
} }
// Ensure that we atomically store prev events AND commit them. If we don't wrap StorePreviousEvents if err = updater.StorePreviousEvents(eventNID, prevEvents); err != nil {
// and EndTransaction in a writer then it's possible for a new write txn to be made between the two return 0, 0, types.StateAtEvent{}, nil, "", fmt.Errorf("updater.StorePreviousEvents: %w", err)
// function calls which will then fail with 'database is locked'. This new write txn would HAVE to be
// something like SetRoomAlias/RemoveRoomAlias as normal input events are already done sequentially due to
// SupportsConcurrentRoomInputs() == false on sqlite, though this does not apply to setting room aliases
// as they don't go via InputRoomEvents
err = d.Writer.Do(d.DB, updater.txn, func(txn *sql.Tx) error {
if err = updater.StorePreviousEvents(eventNID, prevEvents); err != nil {
return fmt.Errorf("updater.StorePreviousEvents: %w", err)
}
succeeded = true
return nil
})
if err != nil {
return 0, 0, types.StateAtEvent{}, nil, "", err
} }
succeeded = true
} }
return eventNID, roomNID, types.StateAtEvent{ return eventNID, roomNID, types.StateAtEvent{