Limit presence in /sync responses (#2394)

* Use filter and limit presence count

* More limiting

* More limiting

* Fix unit test

* Also limit presence by last_active_ts

* Update query, use "from" as the initial lastPos

* Get 1000 presence events, they are filtered later

Co-authored-by: Neil Alexander <neilalexander@users.noreply.github.com>

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)
 }

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)
-
+	afterTS := gomatrixserverlib.AsTimestamp(time.Now().Add(time.Minute * -5))
-	rows, err := stmt.QueryContext(ctx, after)
+	rows, err := stmt.QueryContext(ctx, after, afterTS, filter.Limit)
 	if err != nil {
 		return nil, err
 	}

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) {
+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)
+	return s.Presence.GetPresenceAfter(ctx, nil, after, filter)
 }
 
 func (s *Database) MaxStreamPositionForPresence(ctx context.Context) (types.StreamPosition, error) {

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)
-
+	afterTS := gomatrixserverlib.AsTimestamp(time.Now().Add(time.Minute * -5))
-	rows, err := stmt.QueryContext(ctx, after)
+	rows, err := stmt.QueryContext(ctx, after, afterTS, filter.Limit)
 	if err != nil {
 		return nil, err
 	}

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)
 }

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)
 	}
 

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
 }