From 600fbae31f1b0b6919bb9045958b932071de6737 Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Tue, 22 Feb 2022 13:35:06 +0000 Subject: [PATCH 1/9] Only emit key change notifications from federation when changes are made (#2217) * Only emit key changes when poked over federation * Remove logging * Fix unit test possibly --- keyserver/internal/device_list_update.go | 27 ++++++++++++++++++++---- 1 file changed, 23 insertions(+), 4 deletions(-) diff --git a/keyserver/internal/device_list_update.go b/keyserver/internal/device_list_update.go index c5a5d40c7..b208f0ce5 100644 --- a/keyserver/internal/device_list_update.go +++ b/keyserver/internal/device_list_update.go @@ -241,14 +241,33 @@ func (u *DeviceListUpdater) update(ctx context.Context, event gomatrixserverlib. StreamID: event.StreamID, }, } + + // DeviceKeysJSON will side-effect modify this, so it needs + // to be a copy, not sharing any pointers with the above. + deviceKeysCopy := *keys[0].DeviceKeys + deviceKeysCopy.KeyJSON = nil + existingKeys := []api.DeviceMessage{ + { + Type: keys[0].Type, + DeviceKeys: &deviceKeysCopy, + StreamID: keys[0].StreamID, + }, + } + + // fetch what keys we had already and only emit changes + if err = u.db.DeviceKeysJSON(ctx, existingKeys); err != nil { + // non-fatal, log and continue + util.GetLogger(ctx).WithError(err).WithField("user_id", event.UserID).Errorf( + "failed to query device keys json for calculating diffs", + ) + } + err = u.db.StoreRemoteDeviceKeys(ctx, keys, nil) if err != nil { return false, fmt.Errorf("failed to store remote device keys for %s (%s): %w", event.UserID, event.DeviceID, err) } - // ALWAYS emit key changes when we've been poked over federation even if there's no change - // just in case this poke is important for something. - err = u.producer.ProduceKeyChanges(keys) - if err != nil { + + if err = emitDeviceKeyChanges(u.producer, existingKeys, keys); err != nil { return false, fmt.Errorf("failed to produce device key changes for %s (%s): %w", event.UserID, event.DeviceID, err) } return false, nil From e9545dc12fc699b237f51c1aa67dbbef898b27ce Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Tue, 22 Feb 2022 13:40:08 +0000 Subject: [PATCH 2/9] Remove error when state keys are missing for user NIDs (#2213) * Remove error when state keys are missing for user NIDs There is still an actual bug here somewhere in the membership updater, but this check does more harm than good, since it means that the key consumers don't actually distribute updates to *anyone*. It's better just to deal with this silently for now. To find these broken rows: ``` SELECT * FROM roomserver_membership AS m WHERE NOT EXISTS ( SELECT event_state_key_nid FROM roomserver_event_state_keys AS s WHERE m.sender_nid = s.event_state_key_nid ); ``` * Logging --- roomserver/storage/shared/storage.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/roomserver/storage/shared/storage.go b/roomserver/storage/shared/storage.go index b255cfb3f..e270e121c 100644 --- a/roomserver/storage/shared/storage.go +++ b/roomserver/storage/shared/storage.go @@ -13,6 +13,7 @@ import ( "github.com/matrix-org/dendrite/roomserver/types" "github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/util" + "github.com/sirupsen/logrus" "github.com/tidwall/gjson" ) @@ -1101,7 +1102,7 @@ func (d *Database) JoinedUsersSetInRooms(ctx context.Context, roomIDs []string) return nil, err } if len(nidToUserID) != len(userNIDToCount) { - return nil, fmt.Errorf("found %d users but only have state key nids for %d of them", len(userNIDToCount), len(nidToUserID)) + logrus.Warnf("SelectJoinedUsersSetForRooms found %d users but BulkSelectEventStateKey only returned state key NIDs for %d of them", len(userNIDToCount), len(nidToUserID)) } result := make(map[string]int, len(userNIDToCount)) for nid, count := range userNIDToCount { From 34116178e8ef75569a55655e3001a51f77dfb789 Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Tue, 22 Feb 2022 13:47:14 +0000 Subject: [PATCH 3/9] Remove logging line in `PerformInvite` --- federationapi/internal/perform.go | 1 - 1 file changed, 1 deletion(-) diff --git a/federationapi/internal/perform.go b/federationapi/internal/perform.go index c51ecf146..b888b3654 100644 --- a/federationapi/internal/perform.go +++ b/federationapi/internal/perform.go @@ -554,7 +554,6 @@ func (r *FederationInternalAPI) PerformInvite( if err != nil { return fmt.Errorf("r.federation.SendInviteV2: failed to send invite: %w", err) } - logrus.Infof("GOT INVITE RESPONSE %s", string(inviteRes.Event)) inviteEvent, err := inviteRes.Event.UntrustedEvent(request.RoomVersion) if err != nil { From c7811e9d71041527128f4a1782e2b92453f0f57c Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Tue, 22 Feb 2022 15:43:17 +0000 Subject: [PATCH 4/9] Add `DeviceKeysEqual` (#2219) * Add `DeviceKeysEqual` * Update check order * Fix check * Tweak conditions again * One more time * Single return value --- keyserver/api/api.go | 21 +++++++++++++++++++++ keyserver/internal/internal.go | 2 +- 2 files changed, 22 insertions(+), 1 deletion(-) diff --git a/keyserver/api/api.go b/keyserver/api/api.go index 3933961c1..54eb04f8a 100644 --- a/keyserver/api/api.go +++ b/keyserver/api/api.go @@ -15,6 +15,7 @@ package api import ( + "bytes" "context" "encoding/json" "strings" @@ -73,6 +74,26 @@ type DeviceMessage struct { DeviceChangeID int64 } +// DeviceKeysEqual returns true if the device keys updates contain the +// same display name and key JSON. This will return false if either of +// the updates is not a device keys update, or if the user ID/device ID +// differ between the two. +func (m1 *DeviceMessage) DeviceKeysEqual(m2 *DeviceMessage) bool { + if m1.DeviceKeys == nil || m2.DeviceKeys == nil { + return false + } + if m1.UserID != m2.UserID || m1.DeviceID != m2.DeviceID { + return false + } + if m1.DisplayName != m2.DisplayName { + return false // different display names + } + if len(m1.KeyJSON) == 0 || len(m2.KeyJSON) == 0 { + return false // either is empty + } + return bytes.Equal(m1.KeyJSON, m2.KeyJSON) +} + // DeviceKeys represents a set of device keys for a single device // https://matrix.org/docs/spec/client_server/r0.6.1#post-matrix-client-r0-keys-upload type DeviceKeys struct { diff --git a/keyserver/internal/internal.go b/keyserver/internal/internal.go index 0c264b718..dc3c404bd 100644 --- a/keyserver/internal/internal.go +++ b/keyserver/internal/internal.go @@ -718,7 +718,7 @@ func emitDeviceKeyChanges(producer KeyChangeProducer, existing, new []api.Device for _, existingKey := range existing { // Do not treat the absence of keys as equal, or else we will not emit key changes // when users delete devices which never had a key to begin with as both KeyJSONs are nil. - if bytes.Equal(existingKey.KeyJSON, newKey.KeyJSON) && len(existingKey.KeyJSON) > 0 { + if existingKey.DeviceKeysEqual(&newKey) { exists = true break } From cfff1b0aaafc68efbac27f366360667983f7696a Mon Sep 17 00:00:00 2001 From: S7evinK <2353100+S7evinK@users.noreply.github.com> Date: Tue, 22 Feb 2022 16:50:04 +0100 Subject: [PATCH 5/9] Remote banned user is kicked and may not rejoin until unbanned (#2216) * Remote banned user is kicked and may not rejoin until unbanned * Use gmsl constant --- clientapi/routing/state.go | 2 +- sytest-whitelist | 3 ++- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/clientapi/routing/state.go b/clientapi/routing/state.go index 088e412c6..d25ee8237 100644 --- a/clientapi/routing/state.go +++ b/clientapi/routing/state.go @@ -235,7 +235,7 @@ func OnIncomingStateTypeRequest( } // If the user has never been in the room then stop at this point. // We won't tell the user about a room they have never joined. - if !membershipRes.HasBeenInRoom { + if !membershipRes.HasBeenInRoom || membershipRes.Membership == gomatrixserverlib.Ban { return util.JSONResponse{ Code: http.StatusForbidden, JSON: jsonerror.Forbidden(fmt.Sprintf("Unknown room %q or user %q has never joined this room", roomID, device.UserID)), diff --git a/sytest-whitelist b/sytest-whitelist index 187a0f475..87904938c 100644 --- a/sytest-whitelist +++ b/sytest-whitelist @@ -596,4 +596,5 @@ Device list doesn't change if remote server is down /context/ on joined room works /context/ on non world readable room does not work /context/ returns correct number of events -/context/ with lazy_load_members filter works \ No newline at end of file +/context/ with lazy_load_members filter works +Remote banned user is kicked and may not rejoin until unbanned From 41dc651b25019b9ecc2338ff2ebec202066aaf21 Mon Sep 17 00:00:00 2001 From: S7evinK <2353100+S7evinK@users.noreply.github.com> Date: Tue, 22 Feb 2022 17:34:53 +0100 Subject: [PATCH 6/9] Send device update to local users if remote display name changes (#2215) * Send device_list update to satisfy sytest * Fix build issue from merged in change Co-authored-by: Neil Alexander --- keyserver/internal/device_list_update.go | 6 +++--- keyserver/internal/internal.go | 8 ++++++-- sytest-whitelist | 3 +++ 3 files changed, 12 insertions(+), 5 deletions(-) diff --git a/keyserver/internal/device_list_update.go b/keyserver/internal/device_list_update.go index b208f0ce5..974d0196b 100644 --- a/keyserver/internal/device_list_update.go +++ b/keyserver/internal/device_list_update.go @@ -224,7 +224,7 @@ func (u *DeviceListUpdater) update(ctx context.Context, event gomatrixserverlib. }).Info("DeviceListUpdater.Update") // if we haven't missed anything update the database and notify users - if exists { + if exists || event.Deleted { k := event.Keys if event.Deleted { k = nil @@ -267,7 +267,7 @@ func (u *DeviceListUpdater) update(ctx context.Context, event gomatrixserverlib. return false, fmt.Errorf("failed to store remote device keys for %s (%s): %w", event.UserID, event.DeviceID, err) } - if err = emitDeviceKeyChanges(u.producer, existingKeys, keys); err != nil { + if err = emitDeviceKeyChanges(u.producer, existingKeys, keys, false); err != nil { return false, fmt.Errorf("failed to produce device key changes for %s (%s): %w", event.UserID, event.DeviceID, err) } return false, nil @@ -473,7 +473,7 @@ func (u *DeviceListUpdater) updateDeviceList(res *gomatrixserverlib.RespUserDevi if err != nil { return fmt.Errorf("failed to mark device list as fresh: %w", err) } - err = emitDeviceKeyChanges(u.producer, existingKeys, keys) + err = emitDeviceKeyChanges(u.producer, existingKeys, keys, false) if err != nil { return fmt.Errorf("failed to emit key changes for fresh device list: %w", err) } diff --git a/keyserver/internal/internal.go b/keyserver/internal/internal.go index dc3c404bd..0a8bef95d 100644 --- a/keyserver/internal/internal.go +++ b/keyserver/internal/internal.go @@ -648,7 +648,7 @@ func (a *KeyInternalAPI) uploadLocalDeviceKeys(ctx context.Context, req *api.Per } return } - err = emitDeviceKeyChanges(a.Producer, existingKeys, keysToStore) + err = emitDeviceKeyChanges(a.Producer, existingKeys, keysToStore, req.OnlyDisplayNameUpdates) if err != nil { util.GetLogger(ctx).Errorf("Failed to emitDeviceKeyChanges: %s", err) } @@ -710,7 +710,11 @@ func (a *KeyInternalAPI) uploadOneTimeKeys(ctx context.Context, req *api.Perform } -func emitDeviceKeyChanges(producer KeyChangeProducer, existing, new []api.DeviceMessage) error { +func emitDeviceKeyChanges(producer KeyChangeProducer, existing, new []api.DeviceMessage, onlyUpdateDisplayName bool) error { + // if we only want to update the display names, we can skip the checks below + if onlyUpdateDisplayName { + return producer.ProduceKeyChanges(new) + } // find keys in new that are not in existing var keysAdded []api.DeviceMessage for _, newKey := range new { diff --git a/sytest-whitelist b/sytest-whitelist index 87904938c..d3144572d 100644 --- a/sytest-whitelist +++ b/sytest-whitelist @@ -597,4 +597,7 @@ Device list doesn't change if remote server is down /context/ on non world readable room does not work /context/ returns correct number of events /context/ with lazy_load_members filter works +Can query remote device keys using POST after notification +Device deletion propagates over federation +Get left notifs in sync and /keys/changes when other user leaves Remote banned user is kicked and may not rejoin until unbanned From 2b0a5adfafa46eb4df1ddb7ba909801366b920cf Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Tue, 22 Feb 2022 16:55:08 +0000 Subject: [PATCH 7/9] Version 0.6.4 (#2212) * Version 0.6.4 * Tweaks * Update changelog * Update changelog one last time --- CHANGES.md | 34 ++++++++++++++++++++++++++++++++++ internal/version.go | 2 +- 2 files changed, 35 insertions(+), 1 deletion(-) diff --git a/CHANGES.md b/CHANGES.md index 4df8e869a..ee608194d 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -1,5 +1,39 @@ # Changelog +## Dendrite 0.6.4 (2022-02-21) + +### Features + +* All Client-Server API endpoints are now available under the `/v3` namespace +* The `/whoami` response format now matches the latest Matrix spec version +* Support added for the `/context` endpoint, which should help clients to render quote-replies correctly +* Accounts now have an optional account type field, allowing admin accounts to be created +* Server notices are now supported +* Refactored the user API storage to deduplicate a significant amount of code, as well as merging both user API databases into a single database + * The account database is now used for all user API storage and the device database is now obsolete + * For some installations that have separate account and device databases, this may result in access tokens being revoked and client sessions being logged out — users may need to log in again + * The above can be avoided by moving the `device_devices` table into the account database manually +* Guest registration can now be separately disabled with the new `client_api.guests_disabled` configuration option +* Outbound connections now obey proxy settings from the environment, deprecating the `federation_api.proxy_outbound` configuration options + +### Fixes + +* The roomserver input API will now strictly consume only one database transaction per room, which should prevent situations where the roomserver can deadlock waiting for database connections to become available +* Room joins will now fall back to federation if the local room state is insufficient to create a membership event +* Create events are now correctly filtered from federation `/send` transactions +* Excessive logging when federation is disabled should now be fixed +* Dendrite will no longer panic if trying to retire an invite event that has not been seen yet +* The device list updater will now wait for longer after a connection issue, rather than flooding the logs with errors +* The device list updater will no longer produce unnecessary output events for federated key updates with no changes, which should help to reduce CPU usage +* Local device name changes will now generate key change events correctly +* The sync API will now try to share device list update notifications even if all state key NIDs cannot be fetched +* An off-by-one error in the sync stream token handling which could result in a crash has been fixed +* State events will no longer be re-sent unnecessary by the roomserver to other components if they have already been sent, which should help to reduce the NATS message sizes on the roomserver output topic in some cases +* The roomserver input API now uses the process context and should handle graceful shutdowns better +* Guest registration is now correctly disabled when the `client_api.registration_disabled` configuration option is set +* One-time encryption keys are now cleaned up correctly when a device is logged out or removed +* Invalid state snapshots in the state storage refactoring migration are now reset rather than causing a panic at startup + ## Dendrite 0.6.3 (2022-02-10) ### Features diff --git a/internal/version.go b/internal/version.go index a07f01b61..2ea1c5201 100644 --- a/internal/version.go +++ b/internal/version.go @@ -17,7 +17,7 @@ var build string const ( VersionMajor = 0 VersionMinor = 6 - VersionPatch = 3 + VersionPatch = 4 VersionTag = "" // example: "rc1" ) From b8a97b6ee083d1047d09f99daf2594d0821705ca Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Wed, 23 Feb 2022 10:45:07 +0000 Subject: [PATCH 8/9] Update to matrix-org/pinecone@0f0afd1a46aabf7b1e48cb607bc9fa6a083aade6 --- go.mod | 2 +- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index 2316096df..b104b4192 100644 --- a/go.mod +++ b/go.mod @@ -40,7 +40,7 @@ require ( github.com/matrix-org/go-sqlite3-js v0.0.0-20210709140738-b0d1ba599a6d github.com/matrix-org/gomatrix v0.0.0-20210324163249-be2af5ef2e16 github.com/matrix-org/gomatrixserverlib v0.0.0-20220214133635-20632dd262ed - github.com/matrix-org/pinecone v0.0.0-20220121094951-351265543ddf + github.com/matrix-org/pinecone v0.0.0-20220223104432-0f0afd1a46aa github.com/matrix-org/util v0.0.0-20200807132607-55161520e1d4 github.com/mattn/go-sqlite3 v1.14.10 github.com/morikuni/aec v1.0.0 // indirect diff --git a/go.sum b/go.sum index e79015e51..23ffe8b85 100644 --- a/go.sum +++ b/go.sum @@ -985,8 +985,8 @@ github.com/matrix-org/gomatrix v0.0.0-20210324163249-be2af5ef2e16 h1:ZtO5uywdd5d github.com/matrix-org/gomatrix v0.0.0-20210324163249-be2af5ef2e16/go.mod h1:/gBX06Kw0exX1HrwmoBibFA98yBk/jxKpGVeyQbff+s= github.com/matrix-org/gomatrixserverlib v0.0.0-20220214133635-20632dd262ed h1:R8EiLWArq7KT96DrUq1xq9scPh8vLwKKeCTnORPyjhU= github.com/matrix-org/gomatrixserverlib v0.0.0-20220214133635-20632dd262ed/go.mod h1:qFvhfbQ5orQxlH9vCiFnP4dW27xxnWHdNUBKyj/fbiY= -github.com/matrix-org/pinecone v0.0.0-20220121094951-351265543ddf h1:/nqfHUdQHr3WVdbZieaYFvHF1rin5pvDTa/NOZ/qCyE= -github.com/matrix-org/pinecone v0.0.0-20220121094951-351265543ddf/go.mod h1:r6dsL+ylE0yXe/7zh8y/Bdh6aBYI1r+u4yZni9A4iyk= +github.com/matrix-org/pinecone v0.0.0-20220223104432-0f0afd1a46aa h1:rMYFNVto66gp+eWS8XAUzgp4m0qmUBid6l1HX3mHstk= +github.com/matrix-org/pinecone v0.0.0-20220223104432-0f0afd1a46aa/go.mod h1:r6dsL+ylE0yXe/7zh8y/Bdh6aBYI1r+u4yZni9A4iyk= github.com/matrix-org/util v0.0.0-20190711121626-527ce5ddefc7/go.mod h1:vVQlW/emklohkZnOPwD3LrZUBqdfsbiyO3p1lNV8F6U= github.com/matrix-org/util v0.0.0-20200807132607-55161520e1d4 h1:eCEHXWDv9Rm335MSuB49mFUK44bwZPFSDde3ORE3syk= github.com/matrix-org/util v0.0.0-20200807132607-55161520e1d4/go.mod h1:vVQlW/emklohkZnOPwD3LrZUBqdfsbiyO3p1lNV8F6U= From fea8d152e7d81a9cbd387108d263950d8ec6ddc7 Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Wed, 23 Feb 2022 15:41:32 +0000 Subject: [PATCH 9/9] Relax roomserver input transactional isolation (#2224) * Don't force full transactional isolation on roomserver input * Set succeeded * Tweak `MissingAuthPrevEvents` --- roomserver/internal/helpers/auth.go | 11 +-- roomserver/internal/input/input.go | 57 +----------- roomserver/internal/input/input_events.go | 89 ++++++++++--------- .../internal/input/input_latest_events.go | 11 ++- roomserver/internal/input/input_missing.go | 20 +++-- roomserver/storage/interface.go | 5 ++ roomserver/storage/shared/room_updater.go | 54 ----------- roomserver/storage/shared/storage.go | 23 +++++ 8 files changed, 105 insertions(+), 165 deletions(-) diff --git a/roomserver/internal/helpers/auth.go b/roomserver/internal/helpers/auth.go index 9af0bf591..0229f822f 100644 --- a/roomserver/internal/helpers/auth.go +++ b/roomserver/internal/helpers/auth.go @@ -20,22 +20,17 @@ import ( "sort" "github.com/matrix-org/dendrite/roomserver/state" + "github.com/matrix-org/dendrite/roomserver/storage" "github.com/matrix-org/dendrite/roomserver/types" "github.com/matrix-org/gomatrixserverlib" ) -type checkForAuthAndSoftFailStorage interface { - state.StateResolutionStorage - StateEntriesForEventIDs(ctx context.Context, eventIDs []string) ([]types.StateEntry, error) - RoomInfo(ctx context.Context, roomID string) (*types.RoomInfo, error) -} - // CheckForSoftFail returns true if the event should be soft-failed // and false otherwise. The return error value should be checked before // the soft-fail bool. func CheckForSoftFail( ctx context.Context, - db checkForAuthAndSoftFailStorage, + db storage.Database, event *gomatrixserverlib.HeaderedEvent, stateEventIDs []string, ) (bool, error) { @@ -97,7 +92,7 @@ func CheckForSoftFail( // Returns the numeric IDs for the auth events. func CheckAuthEvents( ctx context.Context, - db checkForAuthAndSoftFailStorage, + db storage.Database, event *gomatrixserverlib.HeaderedEvent, authEventIDs []string, ) ([]types.EventNID, error) { diff --git a/roomserver/internal/input/input.go b/roomserver/internal/input/input.go index 22e4b67a0..178533ded 100644 --- a/roomserver/internal/input/input.go +++ b/roomserver/internal/input/input.go @@ -19,7 +19,6 @@ import ( "context" "encoding/json" "errors" - "fmt" "sync" "time" @@ -40,19 +39,6 @@ import ( "github.com/tidwall/gjson" ) -type retryAction int -type commitAction int - -const ( - doNotRetry retryAction = iota - retryLater -) - -const ( - commitTransaction commitAction = iota - rollbackTransaction -) - var keyContentFields = map[string]string{ "m.room.join_rules": "join_rule", "m.room.history_visibility": "history_visibility", @@ -117,8 +103,7 @@ func (r *Inputer) Start() error { _ = msg.InProgress() // resets the acknowledgement wait timer defer eventsInProgress.Delete(index) defer roomserverInputBackpressure.With(prometheus.Labels{"room_id": roomID}).Dec() - action, err := r.processRoomEventUsingUpdater(r.ProcessContext.Context(), roomID, &inputRoomEvent) - if err != nil { + if err := r.processRoomEvent(r.ProcessContext.Context(), &inputRoomEvent); err != nil { if !errors.Is(err, context.DeadlineExceeded) && !errors.Is(err, context.Canceled) { sentry.CaptureException(err) } @@ -127,11 +112,8 @@ func (r *Inputer) Start() error { "event_id": inputRoomEvent.Event.EventID(), "type": inputRoomEvent.Event.Type(), }).Warn("Roomserver failed to process async event") - } - switch action { - case retryLater: - _ = msg.Nak() - case doNotRetry: + _ = msg.Term() + } else { _ = msg.Ack() } }) @@ -153,37 +135,6 @@ func (r *Inputer) Start() error { return err } -// processRoomEventUsingUpdater opens up a room updater and tries to -// process the event. It returns whether or not we should positively -// or negatively acknowledge the event (i.e. for NATS) and an error -// if it occurred. -func (r *Inputer) processRoomEventUsingUpdater( - ctx context.Context, - roomID string, - inputRoomEvent *api.InputRoomEvent, -) (retryAction, error) { - roomInfo, err := r.DB.RoomInfo(ctx, roomID) - if err != nil { - return doNotRetry, fmt.Errorf("r.DB.RoomInfo: %w", err) - } - updater, err := r.DB.GetRoomUpdater(ctx, roomInfo) - if err != nil { - return retryLater, fmt.Errorf("r.DB.GetRoomUpdater: %w", err) - } - action, err := r.processRoomEvent(ctx, updater, inputRoomEvent) - switch action { - case commitTransaction: - if cerr := updater.Commit(); cerr != nil { - return retryLater, fmt.Errorf("updater.Commit: %w", cerr) - } - case rollbackTransaction: - if rerr := updater.Rollback(); rerr != nil { - return retryLater, fmt.Errorf("updater.Rollback: %w", rerr) - } - } - return doNotRetry, err -} - // InputRoomEvents implements api.RoomserverInternalAPI func (r *Inputer) InputRoomEvents( ctx context.Context, @@ -230,7 +181,7 @@ func (r *Inputer) InputRoomEvents( worker.Act(nil, func() { defer eventsInProgress.Delete(index) defer roomserverInputBackpressure.With(prometheus.Labels{"room_id": roomID}).Dec() - _, err := r.processRoomEventUsingUpdater(ctx, roomID, &inputRoomEvent) + err := r.processRoomEvent(ctx, &inputRoomEvent) if err != nil { if !errors.Is(err, context.DeadlineExceeded) && !errors.Is(err, context.Canceled) { sentry.CaptureException(err) diff --git a/roomserver/internal/input/input_events.go b/roomserver/internal/input/input_events.go index 4e151699e..531d6959e 100644 --- a/roomserver/internal/input/input_events.go +++ b/roomserver/internal/input/input_events.go @@ -26,10 +26,10 @@ import ( "github.com/matrix-org/dendrite/internal" "github.com/matrix-org/dendrite/internal/eventutil" "github.com/matrix-org/dendrite/internal/hooks" + "github.com/matrix-org/dendrite/internal/sqlutil" "github.com/matrix-org/dendrite/roomserver/api" "github.com/matrix-org/dendrite/roomserver/internal/helpers" "github.com/matrix-org/dendrite/roomserver/state" - "github.com/matrix-org/dendrite/roomserver/storage/shared" "github.com/matrix-org/dendrite/roomserver/types" "github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/util" @@ -68,15 +68,14 @@ var processRoomEventDuration = prometheus.NewHistogramVec( // nolint:gocyclo func (r *Inputer) processRoomEvent( ctx context.Context, - updater *shared.RoomUpdater, input *api.InputRoomEvent, -) (commitAction, error) { +) error { select { case <-ctx.Done(): // Before we do anything, make sure the context hasn't expired for this pending task. // If it has then we'll give up straight away — it's probably a synchronous input // request and the caller has already given up, but the inbox task was still queued. - return rollbackTransaction, context.DeadlineExceeded + return context.DeadlineExceeded default: } @@ -109,7 +108,7 @@ func (r *Inputer) processRoomEvent( // if we have already got this event then do not process it again, if the input kind is an outlier. // Outliers contain no extra information which may warrant a re-processing. if input.Kind == api.KindOutlier { - evs, err2 := updater.EventsFromIDs(ctx, []string{event.EventID()}) + evs, err2 := r.DB.EventsFromIDs(ctx, []string{event.EventID()}) if err2 == nil && len(evs) == 1 { // check hash matches if we're on early room versions where the event ID was a random string idFormat, err2 := headered.RoomVersion.EventIDFormat() @@ -118,11 +117,11 @@ func (r *Inputer) processRoomEvent( case gomatrixserverlib.EventIDFormatV1: if bytes.Equal(event.EventReference().EventSHA256, evs[0].EventReference().EventSHA256) { logger.Debugf("Already processed event; ignoring") - return rollbackTransaction, nil + return nil } default: logger.Debugf("Already processed event; ignoring") - return rollbackTransaction, nil + return nil } } } @@ -131,17 +130,21 @@ func (r *Inputer) processRoomEvent( // Don't waste time processing the event if the room doesn't exist. // A room entry locally will only be created in response to a create // event. + roomInfo, rerr := r.DB.RoomInfo(ctx, event.RoomID()) + if rerr != nil { + return fmt.Errorf("r.DB.RoomInfo: %w", rerr) + } isCreateEvent := event.Type() == gomatrixserverlib.MRoomCreate && event.StateKeyEquals("") - if !updater.RoomExists() && !isCreateEvent { - return rollbackTransaction, fmt.Errorf("room %s does not exist for event %s", event.RoomID(), event.EventID()) + if roomInfo == nil && !isCreateEvent { + return fmt.Errorf("room %s does not exist for event %s", event.RoomID(), event.EventID()) } var missingAuth, missingPrev bool serverRes := &fedapi.QueryJoinedHostServerNamesInRoomResponse{} if !isCreateEvent { - missingAuthIDs, missingPrevIDs, err := updater.MissingAuthPrevEvents(ctx, event) + missingAuthIDs, missingPrevIDs, err := r.DB.MissingAuthPrevEvents(ctx, event) if err != nil { - return rollbackTransaction, fmt.Errorf("updater.MissingAuthPrevEvents: %w", err) + return fmt.Errorf("updater.MissingAuthPrevEvents: %w", err) } missingAuth = len(missingAuthIDs) > 0 missingPrev = !input.HasState && len(missingPrevIDs) > 0 @@ -153,7 +156,7 @@ func (r *Inputer) processRoomEvent( ExcludeSelf: true, } if err := r.FSAPI.QueryJoinedHostServerNamesInRoom(ctx, serverReq, serverRes); err != nil { - return rollbackTransaction, fmt.Errorf("r.FSAPI.QueryJoinedHostServerNamesInRoom: %w", err) + return fmt.Errorf("r.FSAPI.QueryJoinedHostServerNamesInRoom: %w", err) } // Sort all of the servers into a map so that we can randomise // their order. Then make sure that the input origin and the @@ -182,8 +185,8 @@ func (r *Inputer) processRoomEvent( isRejected := false authEvents := gomatrixserverlib.NewAuthEvents(nil) knownEvents := map[string]*types.Event{} - if err := r.fetchAuthEvents(ctx, updater, logger, headered, &authEvents, knownEvents, serverRes.ServerNames); err != nil { - return rollbackTransaction, fmt.Errorf("r.fetchAuthEvents: %w", err) + if err := r.fetchAuthEvents(ctx, logger, headered, &authEvents, knownEvents, serverRes.ServerNames); err != nil { + return fmt.Errorf("r.fetchAuthEvents: %w", err) } // Check if the event is allowed by its auth events. If it isn't then @@ -205,12 +208,12 @@ func (r *Inputer) processRoomEvent( // but weren't found. if isRejected { if event.StateKey() != nil { - return commitTransaction, fmt.Errorf( + return fmt.Errorf( "missing auth event %s for state event %s (type %q, state key %q)", authEventID, event.EventID(), event.Type(), *event.StateKey(), ) } else { - return commitTransaction, fmt.Errorf( + return fmt.Errorf( "missing auth event %s for timeline event %s (type %q)", authEventID, event.EventID(), event.Type(), ) @@ -226,7 +229,7 @@ func (r *Inputer) processRoomEvent( // Check that the event passes authentication checks based on the // current room state. var err error - softfail, err = helpers.CheckForSoftFail(ctx, updater, headered, input.StateEventIDs) + softfail, err = helpers.CheckForSoftFail(ctx, r.DB, headered, input.StateEventIDs) if err != nil { logger.WithError(err).Warn("Error authing soft-failed event") } @@ -250,7 +253,8 @@ func (r *Inputer) processRoomEvent( missingState := missingStateReq{ origin: input.Origin, inputer: r, - db: updater, + db: r.DB, + roomInfo: roomInfo, federation: r.FSAPI, keys: r.KeyRing, roomsMu: internal.NewMutexByRoom(), @@ -290,16 +294,16 @@ func (r *Inputer) processRoomEvent( } // Store the event. - _, _, stateAtEvent, redactionEvent, redactedEventID, err := updater.StoreEvent(ctx, event, authEventNIDs, isRejected) + _, _, stateAtEvent, redactionEvent, redactedEventID, err := r.DB.StoreEvent(ctx, event, authEventNIDs, isRejected) if err != nil { - return rollbackTransaction, fmt.Errorf("updater.StoreEvent: %w", err) + return fmt.Errorf("updater.StoreEvent: %w", err) } // if storing this event results in it being redacted then do so. if !isRejected && redactedEventID == event.EventID() { r, rerr := eventutil.RedactEvent(redactionEvent, event) if rerr != nil { - return rollbackTransaction, fmt.Errorf("eventutil.RedactEvent: %w", rerr) + return fmt.Errorf("eventutil.RedactEvent: %w", rerr) } event = r } @@ -310,23 +314,25 @@ func (r *Inputer) processRoomEvent( if input.Kind == api.KindOutlier { logger.Debug("Stored outlier") hooks.Run(hooks.KindNewEventPersisted, headered) - return commitTransaction, nil + return nil } - roomInfo, err := updater.RoomInfo(ctx, event.RoomID()) + // Request the room info again — it's possible that the room has been + // created by now if it didn't exist already. + roomInfo, err = r.DB.RoomInfo(ctx, event.RoomID()) if err != nil { - return rollbackTransaction, fmt.Errorf("updater.RoomInfo: %w", err) + return fmt.Errorf("updater.RoomInfo: %w", err) } if roomInfo == nil { - return rollbackTransaction, fmt.Errorf("updater.RoomInfo missing for room %s", event.RoomID()) + return fmt.Errorf("updater.RoomInfo missing for room %s", event.RoomID()) } if input.HasState || (!missingPrev && stateAtEvent.BeforeStateSnapshotNID == 0) { // We haven't calculated a state for this event yet. // Lets calculate one. - err = r.calculateAndSetState(ctx, updater, input, roomInfo, &stateAtEvent, event, isRejected) + err = r.calculateAndSetState(ctx, input, roomInfo, &stateAtEvent, event, isRejected) if err != nil { - return rollbackTransaction, fmt.Errorf("r.calculateAndSetState: %w", err) + return fmt.Errorf("r.calculateAndSetState: %w", err) } } @@ -337,16 +343,15 @@ func (r *Inputer) processRoomEvent( "missing_prev": missingPrev, }).Warn("Stored rejected event") if rejectionErr != nil { - return commitTransaction, types.RejectedError(rejectionErr.Error()) + return types.RejectedError(rejectionErr.Error()) } - return commitTransaction, nil + return nil } switch input.Kind { case api.KindNew: if err = r.updateLatestEvents( ctx, // context - updater, // room updater roomInfo, // room info for the room being updated stateAtEvent, // state at event (below) event, // event @@ -354,7 +359,7 @@ func (r *Inputer) processRoomEvent( input.TransactionID, // transaction ID input.HasState, // rewrites state? ); err != nil { - return rollbackTransaction, fmt.Errorf("r.updateLatestEvents: %w", err) + return fmt.Errorf("r.updateLatestEvents: %w", err) } case api.KindOld: err = r.WriteOutputEvents(event.RoomID(), []api.OutputEvent{ @@ -366,7 +371,7 @@ func (r *Inputer) processRoomEvent( }, }) if err != nil { - return rollbackTransaction, fmt.Errorf("r.WriteOutputEvents (old): %w", err) + return fmt.Errorf("r.WriteOutputEvents (old): %w", err) } } @@ -385,14 +390,14 @@ func (r *Inputer) processRoomEvent( }, }) if err != nil { - return rollbackTransaction, fmt.Errorf("r.WriteOutputEvents (redactions): %w", err) + return fmt.Errorf("r.WriteOutputEvents (redactions): %w", err) } } // Everything was OK — the latest events updater didn't error and // we've sent output events. Finally, generate a hook call. hooks.Run(hooks.KindNewEventPersisted, headered) - return commitTransaction, nil + return nil } // fetchAuthEvents will check to see if any of the @@ -404,7 +409,6 @@ func (r *Inputer) processRoomEvent( // they are now in the database. func (r *Inputer) fetchAuthEvents( ctx context.Context, - updater *shared.RoomUpdater, logger *logrus.Entry, event *gomatrixserverlib.HeaderedEvent, auth *gomatrixserverlib.AuthEvents, @@ -418,7 +422,7 @@ func (r *Inputer) fetchAuthEvents( } for _, authEventID := range authEventIDs { - authEvents, err := updater.EventsFromIDs(ctx, []string{authEventID}) + authEvents, err := r.DB.EventsFromIDs(ctx, []string{authEventID}) if err != nil || len(authEvents) == 0 || authEvents[0].Event == nil { unknown[authEventID] = struct{}{} continue @@ -495,7 +499,7 @@ nextAuthEvent: } // Finally, store the event in the database. - eventNID, _, _, _, _, err := updater.StoreEvent(ctx, authEvent, authEventNIDs, isRejected) + eventNID, _, _, _, _, err := r.DB.StoreEvent(ctx, authEvent, authEventNIDs, isRejected) if err != nil { return fmt.Errorf("updater.StoreEvent: %w", err) } @@ -520,14 +524,18 @@ nextAuthEvent: func (r *Inputer) calculateAndSetState( ctx context.Context, - updater *shared.RoomUpdater, input *api.InputRoomEvent, roomInfo *types.RoomInfo, stateAtEvent *types.StateAtEvent, event *gomatrixserverlib.Event, isRejected bool, ) error { - var err error + var succeeded bool + updater, err := r.DB.GetRoomUpdater(ctx, roomInfo) + if err != nil { + return fmt.Errorf("r.DB.GetRoomUpdater: %w", err) + } + defer sqlutil.EndTransactionWithCheck(updater, &succeeded, &err) roomState := state.NewStateResolution(updater, roomInfo) if input.HasState { @@ -536,7 +544,7 @@ func (r *Inputer) calculateAndSetState( // We've been told what the state at the event is so we don't need to calculate it. // Check that those state events are in the database and store the state. var entries []types.StateEntry - if entries, err = updater.StateEntriesForEventIDs(ctx, input.StateEventIDs); err != nil { + if entries, err = r.DB.StateEntriesForEventIDs(ctx, input.StateEventIDs); err != nil { return fmt.Errorf("updater.StateEntriesForEventIDs: %w", err) } entries = types.DeduplicateStateEntries(entries) @@ -557,5 +565,6 @@ func (r *Inputer) calculateAndSetState( if err != nil { return fmt.Errorf("r.DB.SetState: %w", err) } + succeeded = true return nil } diff --git a/roomserver/internal/input/input_latest_events.go b/roomserver/internal/input/input_latest_events.go index ae28ebefa..f4a52031a 100644 --- a/roomserver/internal/input/input_latest_events.go +++ b/roomserver/internal/input/input_latest_events.go @@ -20,6 +20,7 @@ import ( "context" "fmt" + "github.com/matrix-org/dendrite/internal/sqlutil" "github.com/matrix-org/dendrite/roomserver/api" "github.com/matrix-org/dendrite/roomserver/state" "github.com/matrix-org/dendrite/roomserver/storage/shared" @@ -47,7 +48,6 @@ import ( // Can only be called once at a time func (r *Inputer) updateLatestEvents( ctx context.Context, - updater *shared.RoomUpdater, roomInfo *types.RoomInfo, stateAtEvent types.StateAtEvent, event *gomatrixserverlib.Event, @@ -55,6 +55,14 @@ func (r *Inputer) updateLatestEvents( transactionID *api.TransactionID, rewritesState bool, ) (err error) { + var succeeded bool + updater, err := r.DB.GetRoomUpdater(ctx, roomInfo) + if err != nil { + return fmt.Errorf("r.DB.GetRoomUpdater: %w", err) + } + + defer sqlutil.EndTransactionWithCheck(updater, &succeeded, &err) + u := latestEventsUpdater{ ctx: ctx, api: r, @@ -71,6 +79,7 @@ func (r *Inputer) updateLatestEvents( return fmt.Errorf("u.doUpdateLatestEvents: %w", err) } + succeeded = true return } diff --git a/roomserver/internal/input/input_missing.go b/roomserver/internal/input/input_missing.go index fc3be7987..4655e92a9 100644 --- a/roomserver/internal/input/input_missing.go +++ b/roomserver/internal/input/input_missing.go @@ -11,7 +11,7 @@ import ( "github.com/matrix-org/dendrite/internal" "github.com/matrix-org/dendrite/roomserver/api" "github.com/matrix-org/dendrite/roomserver/state" - "github.com/matrix-org/dendrite/roomserver/storage/shared" + "github.com/matrix-org/dendrite/roomserver/storage" "github.com/matrix-org/dendrite/roomserver/types" "github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/util" @@ -25,7 +25,8 @@ type parsedRespState struct { type missingStateReq struct { origin gomatrixserverlib.ServerName - db *shared.RoomUpdater + db storage.Database + roomInfo *types.RoomInfo inputer *Inputer keys gomatrixserverlib.JSONVerifier federation fedapi.FederationInternalAPI @@ -80,7 +81,7 @@ func (t *missingStateReq) processEventWithMissingState( // we can just inject all the newEvents as new as we may have only missed 1 or 2 events and have filled // in the gap in the DAG for _, newEvent := range newEvents { - _, err = t.inputer.processRoomEvent(ctx, t.db, &api.InputRoomEvent{ + err = t.inputer.processRoomEvent(ctx, &api.InputRoomEvent{ Kind: api.KindOld, Event: newEvent.Headered(roomVersion), Origin: t.origin, @@ -139,8 +140,7 @@ func (t *missingStateReq) processEventWithMissingState( }) } for _, ire := range outlierRoomEvents { - _, err = t.inputer.processRoomEvent(ctx, t.db, &ire) - if err != nil { + if err = t.inputer.processRoomEvent(ctx, &ire); err != nil { if _, ok := err.(types.RejectedError); !ok { return fmt.Errorf("t.inputer.processRoomEvent (outlier): %w", err) } @@ -163,7 +163,7 @@ func (t *missingStateReq) processEventWithMissingState( stateIDs = append(stateIDs, event.EventID()) } - _, err = t.inputer.processRoomEvent(ctx, t.db, &api.InputRoomEvent{ + err = t.inputer.processRoomEvent(ctx, &api.InputRoomEvent{ Kind: api.KindOld, Event: backwardsExtremity.Headered(roomVersion), Origin: t.origin, @@ -182,7 +182,7 @@ func (t *missingStateReq) processEventWithMissingState( // they will automatically fast-forward based on the room state at the // extremity in the last step. for _, newEvent := range newEvents { - _, err = t.inputer.processRoomEvent(ctx, t.db, &api.InputRoomEvent{ + err = t.inputer.processRoomEvent(ctx, &api.InputRoomEvent{ Kind: api.KindOld, Event: newEvent.Headered(roomVersion), Origin: t.origin, @@ -473,8 +473,10 @@ retryAllowedState: // without `e`. If `isGapFilled=false` then `newEvents` contains the response to /get_missing_events func (t *missingStateReq) getMissingEvents(ctx context.Context, e *gomatrixserverlib.Event, roomVersion gomatrixserverlib.RoomVersion) (newEvents []*gomatrixserverlib.Event, isGapFilled, prevStateKnown bool, err error) { logger := util.GetLogger(ctx).WithField("event_id", e.EventID()).WithField("room_id", e.RoomID()) - - latest := t.db.LatestEvents() + latest, _, _, err := t.db.LatestEventIDs(ctx, t.roomInfo.RoomNID) + if err != nil { + return nil, false, false, fmt.Errorf("t.DB.LatestEventIDs: %w", err) + } latestEvents := make([]string, len(latest)) for i, ev := range latest { latestEvents[i] = ev.EventID diff --git a/roomserver/storage/interface.go b/roomserver/storage/interface.go index a9851e05b..685505d52 100644 --- a/roomserver/storage/interface.go +++ b/roomserver/storage/interface.go @@ -35,6 +35,11 @@ type Database interface { stateBlockNIDs []types.StateBlockNID, state []types.StateEntry, ) (types.StateSnapshotNID, error) + + MissingAuthPrevEvents( + ctx context.Context, e *gomatrixserverlib.Event, + ) (missingAuth, missingPrev []string, err error) + // Look up the state of a room at each event for a list of string event IDs. // Returns an error if there is an error talking to the database. // The length of []types.StateAtEvent is guaranteed to equal the length of eventIDs if no error is returned. diff --git a/roomserver/storage/shared/room_updater.go b/roomserver/storage/shared/room_updater.go index 810a18ef2..d4a2ee3b9 100644 --- a/roomserver/storage/shared/room_updater.go +++ b/roomserver/storage/shared/room_updater.go @@ -103,25 +103,6 @@ func (u *RoomUpdater) CurrentStateSnapshotNID() types.StateSnapshotNID { return u.currentStateSnapshotNID } -func (u *RoomUpdater) MissingAuthPrevEvents( - ctx context.Context, e *gomatrixserverlib.Event, -) (missingAuth, missingPrev []string, err error) { - for _, authEventID := range e.AuthEventIDs() { - if nids, err := u.EventNIDs(ctx, []string{authEventID}); err != nil || len(nids) == 0 { - missingAuth = append(missingAuth, authEventID) - } - } - - for _, prevEventID := range e.PrevEventIDs() { - state, err := u.StateAtEventIDs(ctx, []string{prevEventID}) - if err != nil || len(state) == 0 || (!state[0].IsCreate() && state[0].BeforeStateSnapshotNID == 0) { - missingPrev = append(missingPrev, prevEventID) - } - } - - return -} - // StorePreviousEvents implements types.RoomRecentEventsUpdater - This must be called from a Writer func (u *RoomUpdater) StorePreviousEvents(eventNID types.EventNID, previousEventReferences []gomatrixserverlib.EventReference) error { return u.d.Writer.Do(u.d.DB, u.txn, func(txn *sql.Tx) error { @@ -146,13 +127,6 @@ func (u *RoomUpdater) SnapshotNIDFromEventID( return u.d.snapshotNIDFromEventID(ctx, u.txn, eventID) } -func (u *RoomUpdater) StoreEvent( - ctx context.Context, event *gomatrixserverlib.Event, - authEventNIDs []types.EventNID, isRejected bool, -) (types.EventNID, types.RoomNID, types.StateAtEvent, *gomatrixserverlib.Event, string, error) { - return u.d.storeEvent(ctx, u, event, authEventNIDs, isRejected) -} - func (u *RoomUpdater) StateBlockNIDs( ctx context.Context, stateNIDs []types.StateSnapshotNID, ) ([]types.StateBlockNIDList, error) { @@ -212,44 +186,16 @@ func (u *RoomUpdater) EventIDs( return u.d.EventsTable.BulkSelectEventID(ctx, u.txn, eventNIDs) } -func (u *RoomUpdater) EventNIDs( - ctx context.Context, eventIDs []string, -) (map[string]types.EventNID, error) { - return u.d.eventNIDs(ctx, u.txn, eventIDs, NoFilter) -} - -func (u *RoomUpdater) UnsentEventNIDs( - ctx context.Context, eventIDs []string, -) (map[string]types.EventNID, error) { - return u.d.eventNIDs(ctx, u.txn, eventIDs, FilterUnsentOnly) -} - func (u *RoomUpdater) StateAtEventIDs( ctx context.Context, eventIDs []string, ) ([]types.StateAtEvent, error) { return u.d.EventsTable.BulkSelectStateAtEventByID(ctx, u.txn, eventIDs) } -func (u *RoomUpdater) StateEntriesForEventIDs( - ctx context.Context, eventIDs []string, -) ([]types.StateEntry, error) { - return u.d.EventsTable.BulkSelectStateEventByID(ctx, u.txn, eventIDs) -} - -func (u *RoomUpdater) EventsFromIDs(ctx context.Context, eventIDs []string) ([]types.Event, error) { - return u.d.eventsFromIDs(ctx, u.txn, eventIDs, false) -} - func (u *RoomUpdater) UnsentEventsFromIDs(ctx context.Context, eventIDs []string) ([]types.Event, error) { return u.d.eventsFromIDs(ctx, u.txn, eventIDs, true) } -func (u *RoomUpdater) GetMembershipEventNIDsForRoom( - ctx context.Context, roomNID types.RoomNID, joinOnly bool, localOnly bool, -) ([]types.EventNID, error) { - return u.d.getMembershipEventNIDsForRoom(ctx, u.txn, roomNID, joinOnly, localOnly) -} - // IsReferenced implements types.RoomRecentEventsUpdater func (u *RoomUpdater) IsReferenced(eventReference gomatrixserverlib.EventReference) (bool, error) { err := u.d.PrevEventsTable.SelectPreviousEventExists(u.ctx, u.txn, eventReference.EventID, eventReference.EventSHA256) diff --git a/roomserver/storage/shared/storage.go b/roomserver/storage/shared/storage.go index e270e121c..6e84b2832 100644 --- a/roomserver/storage/shared/storage.go +++ b/roomserver/storage/shared/storage.go @@ -674,6 +674,29 @@ func (d *Database) GetPublishedRooms(ctx context.Context) ([]string, error) { return d.PublishedTable.SelectAllPublishedRooms(ctx, nil, true) } +func (d *Database) MissingAuthPrevEvents( + ctx context.Context, e *gomatrixserverlib.Event, +) (missingAuth, missingPrev []string, err error) { + authEventNIDs, err := d.EventNIDs(ctx, e.AuthEventIDs()) + if err != nil { + return nil, nil, fmt.Errorf("d.EventNIDs: %w", err) + } + for _, authEventID := range e.AuthEventIDs() { + if _, ok := authEventNIDs[authEventID]; !ok { + missingAuth = append(missingAuth, authEventID) + } + } + + for _, prevEventID := range e.PrevEventIDs() { + state, err := d.StateAtEventIDs(ctx, []string{prevEventID}) + if err != nil || len(state) == 0 || (!state[0].IsCreate() && state[0].BeforeStateSnapshotNID == 0) { + missingPrev = append(missingPrev, prevEventID) + } + } + + return +} + func (d *Database) assignRoomNID( ctx context.Context, txn *sql.Tx, roomID string, roomVersion gomatrixserverlib.RoomVersion,