diff --git a/syncapi/storage/interface.go b/syncapi/storage/interface.go index 43aaa358..48697859 100644 --- a/syncapi/storage/interface.go +++ b/syncapi/storage/interface.go @@ -159,6 +159,6 @@ type Database interface { type Presence interface { UpdatePresence(ctx context.Context, userID string, presence types.Presence, statusMsg *string, lastActiveTS gomatrixserverlib.Timestamp, fromSync bool) (types.StreamPosition, error) GetPresence(ctx context.Context, userID string) (*types.PresenceInternal, error) - PresenceAfter(ctx context.Context, after types.StreamPosition) (map[string]*types.PresenceInternal, error) + PresenceAfter(ctx context.Context, after types.StreamPosition, filter gomatrixserverlib.EventFilter) (map[string]*types.PresenceInternal, error) MaxStreamPositionForPresence(ctx context.Context) (types.StreamPosition, error) } diff --git a/syncapi/storage/postgres/presence_table.go b/syncapi/storage/postgres/presence_table.go index 9f1e37f7..7194afea 100644 --- a/syncapi/storage/postgres/presence_table.go +++ b/syncapi/storage/postgres/presence_table.go @@ -17,6 +17,7 @@ package postgres import ( "context" "database/sql" + "time" "github.com/matrix-org/dendrite/internal" "github.com/matrix-org/dendrite/internal/sqlutil" @@ -72,7 +73,8 @@ const selectMaxPresenceSQL = "" + const selectPresenceAfter = "" + " SELECT id, user_id, presence, status_msg, last_active_ts" + " FROM syncapi_presence" + - " WHERE id > $1" + " WHERE id > $1 AND last_active_ts >= $2" + + " ORDER BY id ASC LIMIT $3" type presenceStatements struct { upsertPresenceStmt *sql.Stmt @@ -144,11 +146,12 @@ func (p *presenceStatements) GetMaxPresenceID(ctx context.Context, txn *sql.Tx) func (p *presenceStatements) GetPresenceAfter( ctx context.Context, txn *sql.Tx, after types.StreamPosition, + filter gomatrixserverlib.EventFilter, ) (presences map[string]*types.PresenceInternal, err error) { presences = make(map[string]*types.PresenceInternal) stmt := sqlutil.TxStmt(txn, p.selectPresenceAfterStmt) - - rows, err := stmt.QueryContext(ctx, after) + afterTS := gomatrixserverlib.AsTimestamp(time.Now().Add(time.Minute * -5)) + rows, err := stmt.QueryContext(ctx, after, afterTS, filter.Limit) if err != nil { return nil, err } diff --git a/syncapi/storage/shared/syncserver.go b/syncapi/storage/shared/syncserver.go index 25aca50a..b7d2d3a2 100644 --- a/syncapi/storage/shared/syncserver.go +++ b/syncapi/storage/shared/syncserver.go @@ -1056,8 +1056,8 @@ func (s *Database) GetPresence(ctx context.Context, userID string) (*types.Prese return s.Presence.GetPresenceForUser(ctx, nil, userID) } -func (s *Database) PresenceAfter(ctx context.Context, after types.StreamPosition) (map[string]*types.PresenceInternal, error) { - return s.Presence.GetPresenceAfter(ctx, nil, after) +func (s *Database) PresenceAfter(ctx context.Context, after types.StreamPosition, filter gomatrixserverlib.EventFilter) (map[string]*types.PresenceInternal, error) { + return s.Presence.GetPresenceAfter(ctx, nil, after, filter) } func (s *Database) MaxStreamPositionForPresence(ctx context.Context) (types.StreamPosition, error) { diff --git a/syncapi/storage/sqlite3/presence_table.go b/syncapi/storage/sqlite3/presence_table.go index 177a01bf..b61a825d 100644 --- a/syncapi/storage/sqlite3/presence_table.go +++ b/syncapi/storage/sqlite3/presence_table.go @@ -17,6 +17,7 @@ package sqlite3 import ( "context" "database/sql" + "time" "github.com/matrix-org/dendrite/internal" "github.com/matrix-org/dendrite/internal/sqlutil" @@ -71,7 +72,8 @@ const selectMaxPresenceSQL = "" + const selectPresenceAfter = "" + " SELECT id, user_id, presence, status_msg, last_active_ts" + " FROM syncapi_presence" + - " WHERE id > $1" + " WHERE id > $1 AND last_active_ts >= $2" + + " ORDER BY id ASC LIMIT $3" type presenceStatements struct { db *sql.DB @@ -158,12 +160,12 @@ func (p *presenceStatements) GetMaxPresenceID(ctx context.Context, txn *sql.Tx) // GetPresenceAfter returns the changes presences after a given stream id func (p *presenceStatements) GetPresenceAfter( ctx context.Context, txn *sql.Tx, - after types.StreamPosition, + after types.StreamPosition, filter gomatrixserverlib.EventFilter, ) (presences map[string]*types.PresenceInternal, err error) { presences = make(map[string]*types.PresenceInternal) stmt := sqlutil.TxStmt(txn, p.selectPresenceAfterStmt) - - rows, err := stmt.QueryContext(ctx, after) + afterTS := gomatrixserverlib.AsTimestamp(time.Now().Add(time.Minute * -5)) + rows, err := stmt.QueryContext(ctx, after, afterTS, filter.Limit) if err != nil { return nil, err } diff --git a/syncapi/storage/tables/interface.go b/syncapi/storage/tables/interface.go index 4ff4689e..6fdd9483 100644 --- a/syncapi/storage/tables/interface.go +++ b/syncapi/storage/tables/interface.go @@ -188,5 +188,5 @@ type Presence interface { UpsertPresence(ctx context.Context, txn *sql.Tx, userID string, statusMsg *string, presence types.Presence, lastActiveTS gomatrixserverlib.Timestamp, fromSync bool) (pos types.StreamPosition, err error) GetPresenceForUser(ctx context.Context, txn *sql.Tx, userID string) (presence *types.PresenceInternal, err error) GetMaxPresenceID(ctx context.Context, txn *sql.Tx) (pos types.StreamPosition, err error) - GetPresenceAfter(ctx context.Context, txn *sql.Tx, after types.StreamPosition) (presences map[string]*types.PresenceInternal, err error) + GetPresenceAfter(ctx context.Context, txn *sql.Tx, after types.StreamPosition, filter gomatrixserverlib.EventFilter) (presences map[string]*types.PresenceInternal, err error) } diff --git a/syncapi/streams/stream_presence.go b/syncapi/streams/stream_presence.go index 614b88d4..675a7a17 100644 --- a/syncapi/streams/stream_presence.go +++ b/syncapi/streams/stream_presence.go @@ -53,7 +53,8 @@ func (p *PresenceStreamProvider) IncrementalSync( req *types.SyncRequest, from, to types.StreamPosition, ) types.StreamPosition { - presences, err := p.DB.PresenceAfter(ctx, from) + // We pull out a larger number than the filter asks for, since we're filtering out events later + presences, err := p.DB.PresenceAfter(ctx, from, gomatrixserverlib.EventFilter{Limit: 1000}) if err != nil { req.Log.WithError(err).Error("p.DB.PresenceAfter failed") return from @@ -72,6 +73,7 @@ func (p *PresenceStreamProvider) IncrementalSync( req.Log.WithError(err).Error("unable to refresh notifier lists") return from } + NewlyJoinedLoop: for _, roomID := range newlyJoined { roomUsers := p.notifier.JoinedUsers(roomID) for i := range roomUsers { @@ -86,11 +88,14 @@ func (p *PresenceStreamProvider) IncrementalSync( req.Log.WithError(err).Error("unable to query presence for user") return from } + if len(presences) > req.Filter.Presence.Limit { + break NewlyJoinedLoop + } } } } - lastPos := to + lastPos := from for _, presence := range presences { if presence == nil { continue @@ -135,6 +140,9 @@ func (p *PresenceStreamProvider) IncrementalSync( if presence.StreamPos > lastPos { lastPos = presence.StreamPos } + if len(req.Response.Presence.Events) == req.Filter.Presence.Limit { + break + } p.cache.Store(cacheKey, presence) } diff --git a/syncapi/sync/requestpool_test.go b/syncapi/sync/requestpool_test.go index a8008994..5e52bc7c 100644 --- a/syncapi/sync/requestpool_test.go +++ b/syncapi/sync/requestpool_test.go @@ -30,7 +30,7 @@ func (d dummyDB) GetPresence(ctx context.Context, userID string) (*types.Presenc return &types.PresenceInternal{}, nil } -func (d dummyDB) PresenceAfter(ctx context.Context, after types.StreamPosition) (map[string]*types.PresenceInternal, error) { +func (d dummyDB) PresenceAfter(ctx context.Context, after types.StreamPosition, filter gomatrixserverlib.EventFilter) (map[string]*types.PresenceInternal, error) { return map[string]*types.PresenceInternal{}, nil }