MirrorHub/dendrite
0
0
Fork 0
mirror of https://github.com/matrix-org/dendrite synced 2024-11-12 04:41:11 +01:00
Code Issues Releases Wiki Activity

Don't re-request state events that are already in the timeline (#1739)

Browse source 
* Don't request state events if we already have the timeline events (Postgres only)

* Rename variable

* nocyclo

* Add SQLite

* Tweaks

* Revert query change

* Don't dedupe if asking for full state

* Update query
This commit is contained in:
Neil Alexander 2021-02-04 12:20:37 +00:00 committed by GitHub
parent 6099379ea4
commit 6e44450cc9
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
8 changed files with 45 additions and 23 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
syncapi/storage/interface.go
View file

@ -35,7 +35,7 @@ type Database interface {
MaxStreamPositionForAccountData(ctx context.Context) (types.StreamPosition, error) MaxStreamPositionForAccountData(ctx context.Context) (types.StreamPosition, error)
MaxStreamPositionForSendToDeviceMessages(ctx context.Context) (types.StreamPosition, error) MaxStreamPositionForSendToDeviceMessages(ctx context.Context) (types.StreamPosition, error)
CurrentState(ctx context.Context, roomID string, stateFilterPart *gomatrixserverlib.StateFilter) ([]*gomatrixserverlib.HeaderedEvent, error) CurrentState(ctx context.Context, roomID string, stateFilterPart *gomatrixserverlib.StateFilter, excludeEventIDs []string) ([]*gomatrixserverlib.HeaderedEvent, error)
GetStateDeltasForFullStateSync(ctx context.Context, device *userapi.Device, r types.Range, userID string, stateFilter *gomatrixserverlib.StateFilter) ([]types.StateDelta, []string, error) GetStateDeltasForFullStateSync(ctx context.Context, device *userapi.Device, r types.Range, userID string, stateFilter *gomatrixserverlib.StateFilter) ([]types.StateDelta, []string, error)
GetStateDeltas(ctx context.Context, device *userapi.Device, r types.Range, userID string, stateFilter *gomatrixserverlib.StateFilter) ([]types.StateDelta, []string, error) GetStateDeltas(ctx context.Context, device *userapi.Device, r types.Range, userID string, stateFilter *gomatrixserverlib.StateFilter) ([]types.StateDelta, []string, error)
RoomIDsWithMembership(ctx context.Context, userID string, membership string) ([]string, error) RoomIDsWithMembership(ctx context.Context, userID string, membership string) ([]string, error)

5
syncapi/storage/postgres/current_room_state_table.go
View file

@ -84,7 +84,8 @@ const selectCurrentStateSQL = "" +
" AND ( $4::text[] IS NULL OR type LIKE ANY($4) )" + " AND ( $4::text[] IS NULL OR type LIKE ANY($4) )" +
" AND ( $5::text[] IS NULL OR NOT(type LIKE ANY($5)) )" + " AND ( $5::text[] IS NULL OR NOT(type LIKE ANY($5)) )" +
" AND ( $6::bool IS NULL OR contains_url = $6 )" + " AND ( $6::bool IS NULL OR contains_url = $6 )" +
" LIMIT $7" " AND (event_id = ANY($7)) IS NOT TRUE" +
" LIMIT $8"
const selectJoinedUsersSQL = "" + const selectJoinedUsersSQL = "" +
"SELECT room_id, state_key FROM syncapi_current_room_state WHERE type = 'm.room.member' AND membership = 'join'" "SELECT room_id, state_key FROM syncapi_current_room_state WHERE type = 'm.room.member' AND membership = 'join'"
@ -197,6 +198,7 @@ func (s *currentRoomStateStatements) SelectRoomIDsWithMembership(
func (s *currentRoomStateStatements) SelectCurrentState( func (s *currentRoomStateStatements) SelectCurrentState(
ctx context.Context, txn *sql.Tx, roomID string, ctx context.Context, txn *sql.Tx, roomID string,
stateFilter *gomatrixserverlib.StateFilter, stateFilter *gomatrixserverlib.StateFilter,
excludeEventIDs []string,
) ([]*gomatrixserverlib.HeaderedEvent, error) { ) ([]*gomatrixserverlib.HeaderedEvent, error) {
stmt := sqlutil.TxStmt(txn, s.selectCurrentStateStmt) stmt := sqlutil.TxStmt(txn, s.selectCurrentStateStmt)
rows, err := stmt.QueryContext(ctx, roomID, rows, err := stmt.QueryContext(ctx, roomID,
@ -205,6 +207,7 @@ func (s *currentRoomStateStatements) SelectCurrentState(
pq.StringArray(filterConvertTypeWildcardToSQL(stateFilter.Types)), pq.StringArray(filterConvertTypeWildcardToSQL(stateFilter.Types)),
pq.StringArray(filterConvertTypeWildcardToSQL(stateFilter.NotTypes)), pq.StringArray(filterConvertTypeWildcardToSQL(stateFilter.NotTypes)),
stateFilter.ContainsURL, stateFilter.ContainsURL,
pq.StringArray(excludeEventIDs),
stateFilter.Limit, stateFilter.Limit,
) )
if err != nil { if err != nil {

8
syncapi/storage/shared/syncserver.go
View file

@ -103,8 +103,8 @@ func (d *Database) MaxStreamPositionForAccountData(ctx context.Context) (types.S
return types.StreamPosition(id), nil return types.StreamPosition(id), nil
} }
func (d *Database) CurrentState(ctx context.Context, roomID string, stateFilterPart *gomatrixserverlib.StateFilter) ([]*gomatrixserverlib.HeaderedEvent, error) { func (d *Database) CurrentState(ctx context.Context, roomID string, stateFilterPart *gomatrixserverlib.StateFilter, excludeEventIDs []string) ([]*gomatrixserverlib.HeaderedEvent, error) {
return d.CurrentRoomState.SelectCurrentState(ctx, nil, roomID, stateFilterPart) return d.CurrentRoomState.SelectCurrentState(ctx, nil, roomID, stateFilterPart, excludeEventIDs)
} }
func (d *Database) RoomIDsWithMembership(ctx context.Context, userID string, membership string) ([]string, error) { func (d *Database) RoomIDsWithMembership(ctx context.Context, userID string, membership string) ([]string, error) {
@ -195,7 +195,7 @@ func (d *Database) GetStateEvent(
func (d *Database) GetStateEventsForRoom( func (d *Database) GetStateEventsForRoom(
ctx context.Context, roomID string, stateFilter *gomatrixserverlib.StateFilter, ctx context.Context, roomID string, stateFilter *gomatrixserverlib.StateFilter,
) (stateEvents []*gomatrixserverlib.HeaderedEvent, err error) { ) (stateEvents []*gomatrixserverlib.HeaderedEvent, err error) {
stateEvents, err = d.CurrentRoomState.SelectCurrentState(ctx, nil, roomID, stateFilter) stateEvents, err = d.CurrentRoomState.SelectCurrentState(ctx, nil, roomID, stateFilter, nil)
return return
} }
@ -870,7 +870,7 @@ func (d *Database) currentStateStreamEventsForRoom(
ctx context.Context, txn *sql.Tx, roomID string, ctx context.Context, txn *sql.Tx, roomID string,
stateFilter *gomatrixserverlib.StateFilter, stateFilter *gomatrixserverlib.StateFilter,
) ([]types.StreamEvent, error) { ) ([]types.StreamEvent, error) {
allState, err := d.CurrentRoomState.SelectCurrentState(ctx, txn, roomID, stateFilter) allState, err := d.CurrentRoomState.SelectCurrentState(ctx, txn, roomID, stateFilter, nil)
if err != nil { if err != nil {
return nil, err return nil, err
} }

3
syncapi/storage/sqlite3/current_room_state_table.go
View file

@ -178,6 +178,7 @@ func (s *currentRoomStateStatements) SelectRoomIDsWithMembership(
func (s *currentRoomStateStatements) SelectCurrentState( func (s *currentRoomStateStatements) SelectCurrentState(
ctx context.Context, txn *sql.Tx, roomID string, ctx context.Context, txn *sql.Tx, roomID string,
stateFilter *gomatrixserverlib.StateFilter, stateFilter *gomatrixserverlib.StateFilter,
excludeEventIDs []string,
) ([]*gomatrixserverlib.HeaderedEvent, error) { ) ([]*gomatrixserverlib.HeaderedEvent, error) {
stmt, params, err := prepareWithFilters( stmt, params, err := prepareWithFilters(
s.db, txn, selectCurrentStateSQL, s.db, txn, selectCurrentStateSQL,
@ -186,7 +187,7 @@ func (s *currentRoomStateStatements) SelectCurrentState(
}, },
stateFilter.Senders, stateFilter.NotSenders, stateFilter.Senders, stateFilter.NotSenders,
stateFilter.Types, stateFilter.NotTypes, stateFilter.Types, stateFilter.NotTypes,
stateFilter.Limit, FilterOrderNone, excludeEventIDs, stateFilter.Limit, FilterOrderNone,
) )
if err != nil { if err != nil {
return nil, fmt.Errorf("s.prepareWithFilters: %w", err) return nil, fmt.Errorf("s.prepareWithFilters: %w", err)

9
syncapi/storage/sqlite3/filtering.go
View file

@ -23,9 +23,10 @@ const (
// fields might come from either a StateFilter or an EventFilter, // fields might come from either a StateFilter or an EventFilter,
// and it's easier just to have the caller extract the relevant // and it's easier just to have the caller extract the relevant
// parts. // parts.
// nolint:gocyclo
func prepareWithFilters( func prepareWithFilters(
db *sql.DB, txn *sql.Tx, query string, params []interface{}, db *sql.DB, txn *sql.Tx, query string, params []interface{},
senders, notsenders, types, nottypes []string, senders, notsenders, types, nottypes []string, excludeEventIDs []string,
limit int, order FilterOrder, limit int, order FilterOrder,
) (*sql.Stmt, []interface{}, error) { ) (*sql.Stmt, []interface{}, error) {
offset := len(params) offset := len(params)
@ -53,6 +54,12 @@ func prepareWithFilters(
params, offset = append(params, v), offset+1 params, offset = append(params, v), offset+1
} }
} }
if count := len(excludeEventIDs); count > 0 {
query += " AND event_id NOT IN " + sqlutil.QueryVariadicOffset(count, offset)
for _, v := range excludeEventIDs {
params, offset = append(params, v), offset+1
}
}
switch order { switch order {
case FilterOrderAsc: case FilterOrderAsc:
query += " ORDER BY id ASC" query += " ORDER BY id ASC"

6
syncapi/storage/sqlite3/output_room_events_table.go
View file

@ -150,7 +150,7 @@ func (s *outputRoomEventsStatements) SelectStateInRange(
}, },
stateFilter.Senders, stateFilter.NotSenders, stateFilter.Senders, stateFilter.NotSenders,
stateFilter.Types, stateFilter.NotTypes, stateFilter.Types, stateFilter.NotTypes,
stateFilter.Limit, FilterOrderAsc, nil, stateFilter.Limit, FilterOrderAsc,
) )
if err != nil { if err != nil {
return nil, nil, fmt.Errorf("s.prepareWithFilters: %w", err) return nil, nil, fmt.Errorf("s.prepareWithFilters: %w", err)
@ -326,7 +326,7 @@ func (s *outputRoomEventsStatements) SelectRecentEvents(
}, },
eventFilter.Senders, eventFilter.NotSenders, eventFilter.Senders, eventFilter.NotSenders,
eventFilter.Types, eventFilter.NotTypes, eventFilter.Types, eventFilter.NotTypes,
eventFilter.Limit+1, FilterOrderDesc, nil, eventFilter.Limit+1, FilterOrderDesc,
) )
if err != nil { if err != nil {
return nil, false, fmt.Errorf("s.prepareWithFilters: %w", err) return nil, false, fmt.Errorf("s.prepareWithFilters: %w", err)
@ -374,7 +374,7 @@ func (s *outputRoomEventsStatements) SelectEarlyEvents(
}, },
eventFilter.Senders, eventFilter.NotSenders, eventFilter.Senders, eventFilter.NotSenders,
eventFilter.Types, eventFilter.NotTypes, eventFilter.Types, eventFilter.NotTypes,
eventFilter.Limit, FilterOrderAsc, nil, eventFilter.Limit, FilterOrderAsc,
) )
if err != nil { if err != nil {
return nil, fmt.Errorf("s.prepareWithFilters: %w", err) return nil, fmt.Errorf("s.prepareWithFilters: %w", err)

2
syncapi/storage/tables/interface.go
View file

@ -91,7 +91,7 @@ type CurrentRoomState interface {
DeleteRoomStateByEventID(ctx context.Context, txn *sql.Tx, eventID string) error DeleteRoomStateByEventID(ctx context.Context, txn *sql.Tx, eventID string) error
DeleteRoomStateForRoom(ctx context.Context, txn *sql.Tx, roomID string) error DeleteRoomStateForRoom(ctx context.Context, txn *sql.Tx, roomID string) error
// SelectCurrentState returns all the current state events for the given room. // SelectCurrentState returns all the current state events for the given room.
SelectCurrentState(ctx context.Context, txn *sql.Tx, roomID string, stateFilter *gomatrixserverlib.StateFilter) ([]*gomatrixserverlib.HeaderedEvent, error) SelectCurrentState(ctx context.Context, txn *sql.Tx, roomID string, stateFilter *gomatrixserverlib.StateFilter, excludeEventIDs []string) ([]*gomatrixserverlib.HeaderedEvent, error)
// SelectRoomIDsWithMembership returns the list of room IDs which have the given user in the given membership state. // SelectRoomIDsWithMembership returns the list of room IDs which have the given user in the given membership state.
SelectRoomIDsWithMembership(ctx context.Context, txn *sql.Tx, userID string, membership string) ([]string, error) SelectRoomIDsWithMembership(ctx context.Context, txn *sql.Tx, userID string, membership string) ([]string, error)
// SelectJoinedUsers returns a map of room ID to a list of joined user IDs. // SelectJoinedUsers returns a map of room ID to a list of joined user IDs.

33
syncapi/streams/stream_pdu.go
View file

@ -98,7 +98,7 @@ func (p *PDUStreamProvider) CompleteSync(
var jr *types.JoinResponse var jr *types.JoinResponse
jr, err = p.getJoinResponseForCompleteSync( jr, err = p.getJoinResponseForCompleteSync(
ctx, roomID, r, &stateFilter, &eventFilter, req.Device, ctx, roomID, r, &stateFilter, &eventFilter, req.WantFullState, req.Device,
) )
if err != nil { if err != nil {
req.Log.WithError(err).Error("p.getJoinResponseForCompleteSync failed") req.Log.WithError(err).Error("p.getJoinResponseForCompleteSync failed")
@ -124,7 +124,7 @@ func (p *PDUStreamProvider) CompleteSync(
if !peek.Deleted { if !peek.Deleted {
var jr *types.JoinResponse var jr *types.JoinResponse
jr, err = p.getJoinResponseForCompleteSync( jr, err = p.getJoinResponseForCompleteSync(
ctx, peek.RoomID, r, &stateFilter, &eventFilter, req.Device, ctx, peek.RoomID, r, &stateFilter, &eventFilter, req.WantFullState, req.Device,
) )
if err != nil { if err != nil {
req.Log.WithError(err).Error("p.getJoinResponseForCompleteSync failed") req.Log.WithError(err).Error("p.getJoinResponseForCompleteSync failed")
@ -254,26 +254,37 @@ func (p *PDUStreamProvider) addRoomDeltaToResponse(
return nil return nil
} }
// nolint:gocyclo
func (p *PDUStreamProvider) getJoinResponseForCompleteSync( func (p *PDUStreamProvider) getJoinResponseForCompleteSync(
ctx context.Context, ctx context.Context,
roomID string, roomID string,
r types.Range, r types.Range,
stateFilter *gomatrixserverlib.StateFilter, stateFilter *gomatrixserverlib.StateFilter,
eventFilter *gomatrixserverlib.RoomEventFilter, eventFilter *gomatrixserverlib.RoomEventFilter,
wantFullState bool,
device *userapi.Device, device *userapi.Device,
) (jr *types.JoinResponse, err error) { ) (jr *types.JoinResponse, err error) {
var stateEvents []*gomatrixserverlib.HeaderedEvent // TODO: When filters are added, we may need to call this multiple times to get enough events.
stateEvents, err = p.DB.CurrentState(ctx, roomID, stateFilter) // See: https://github.com/matrix-org/synapse/blob/v0.19.3/synapse/handlers/sync.py#L316
recentStreamEvents, limited, err := p.DB.RecentEvents(
ctx, roomID, r, eventFilter, true, true,
)
if err != nil { if err != nil {
return return
} }
// TODO: When filters are added, we may need to call this multiple times to get enough events.
// See: https://github.com/matrix-org/synapse/blob/v0.19.3/synapse/handlers/sync.py#L316 // Get the event IDs of the stream events we fetched. There's no point in us
var recentStreamEvents []types.StreamEvent var excludingEventIDs []string
var limited bool if !wantFullState {
recentStreamEvents, limited, err = p.DB.RecentEvents( excludingEventIDs = make([]string, 0, len(recentStreamEvents))
ctx, roomID, r, eventFilter, true, true, for _, event := range recentStreamEvents {
) if event.StateKey() != nil {
excludingEventIDs = append(excludingEventIDs, event.EventID())
}
}
}
stateEvents, err := p.DB.CurrentState(ctx, roomID, stateFilter, excludingEventIDs)
if err != nil { if err != nil {
return return
} }