From 0c51a17acdd0ed82b4671ced8b21256273a47207 Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Wed, 18 Mar 2020 20:08:55 +0000 Subject: [PATCH] Implement federated public rooms - Make thirdparty endpoint r0 so riot-web loads the public room list --- clientapi/routing/routing.go | 2 +- cmd/dendritejs/main.go | 2 +- cmd/dendritejs/publicrooms.go | 6 - publicroomsapi/directory/public_rooms.go | 123 +++++++++++++++++- publicroomsapi/publicroomsapi.go | 4 +- publicroomsapi/routing/routing.go | 23 +--- .../storage/sqlite3/public_rooms_table.go | 4 +- publicroomsapi/types/types.go | 5 +- 8 files changed, 136 insertions(+), 33 deletions(-) diff --git a/clientapi/routing/routing.go b/clientapi/routing/routing.go index f0841b796..400a275a9 100644 --- a/clientapi/routing/routing.go +++ b/clientapi/routing/routing.go @@ -390,7 +390,7 @@ func Setup( }), ).Methods(http.MethodGet, http.MethodOptions) - unstableMux.Handle("/thirdparty/protocols", + r0mux.Handle("/thirdparty/protocols", common.MakeExternalAPI("thirdparty_protocols", func(req *http.Request) util.JSONResponse { // TODO: Return the third party protcols return util.JSONResponse{ diff --git a/cmd/dendritejs/main.go b/cmd/dendritejs/main.go index 13ae7ecbc..7c8526715 100644 --- a/cmd/dendritejs/main.go +++ b/cmd/dendritejs/main.go @@ -135,7 +135,7 @@ func main() { ) federationapi.SetupFederationAPIComponent(base, accountDB, deviceDB, federation, &keyRing, alias, input, query, asQuery, fedSenderAPI) mediaapi.SetupMediaAPIComponent(base, deviceDB) - publicroomsapi.SetupPublicRoomsAPIComponent(base, deviceDB, query, p2pPublicRoomProvider) + publicroomsapi.SetupPublicRoomsAPIComponent(base, deviceDB, query, federation, p2pPublicRoomProvider) syncapi.SetupSyncAPIComponent(base, deviceDB, accountDB, query, federation, cfg) httpHandler := common.WrapHandlerInCORS(base.APIMux) diff --git a/cmd/dendritejs/publicrooms.go b/cmd/dendritejs/publicrooms.go index 930057672..17822e7ad 100644 --- a/cmd/dendritejs/publicrooms.go +++ b/cmd/dendritejs/publicrooms.go @@ -17,8 +17,6 @@ package main import ( - "time" - "github.com/matrix-org/go-http-js-libp2p/go_http_js_libp2p" ) @@ -39,10 +37,6 @@ func (p *libp2pPublicRoomsProvider) foundProviders(peerInfos []go_http_js_libp2p p.providers = peerInfos } -func (p *libp2pPublicRoomsProvider) PollInterval() time.Duration { - return 10 * time.Second -} - func (p *libp2pPublicRoomsProvider) Homeservers() []string { result := make([]string, len(p.providers)) for i := range p.providers { diff --git a/publicroomsapi/directory/public_rooms.go b/publicroomsapi/directory/public_rooms.go index ffef49927..2f883d02c 100644 --- a/publicroomsapi/directory/public_rooms.go +++ b/publicroomsapi/directory/public_rooms.go @@ -18,10 +18,13 @@ import ( "context" "net/http" "strconv" + "sync" + "time" "github.com/matrix-org/dendrite/clientapi/httputil" "github.com/matrix-org/dendrite/clientapi/jsonerror" "github.com/matrix-org/dendrite/publicroomsapi/storage" + "github.com/matrix-org/dendrite/publicroomsapi/types" "github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/util" ) @@ -44,7 +47,7 @@ func GetPostPublicRooms( if fillErr := fillPublicRoomsReq(req, &request); fillErr != nil { return *fillErr } - response, err := PublicRooms(req.Context(), request, publicRoomDatabase) + response, err := publicRooms(req.Context(), request, publicRoomDatabase) if err != nil { return jsonerror.InternalServerError() } @@ -54,7 +57,123 @@ func GetPostPublicRooms( } } -func PublicRooms(ctx context.Context, request PublicRoomReq, publicRoomDatabase storage.Database) (*gomatrixserverlib.RespPublicRooms, error) { +// GetPostPublicRoomsWithExternal is the same as GetPostPublicRooms but also mixes in public rooms from the provider supplied. +func GetPostPublicRoomsWithExternal( + req *http.Request, publicRoomDatabase storage.Database, fedClient *gomatrixserverlib.FederationClient, + extRoomsProvider types.ExternalPublicRoomsProvider, +) util.JSONResponse { + var request PublicRoomReq + if fillErr := fillPublicRoomsReq(req, &request); fillErr != nil { + return *fillErr + } + response, err := publicRooms(req.Context(), request, publicRoomDatabase) + if err != nil { + return jsonerror.InternalServerError() + } + + if request.Since != "" { + // TODO: handle pagination tokens sensibly rather than ignoring them. + // ignore paginated requests since we don't handle them yet over federation. + // Only the initial request will contain federated rooms. + return util.JSONResponse{ + Code: http.StatusOK, + JSON: response, + } + } + + // If we have already hit the limit on the number of rooms, bail. + var limit int + if request.Limit > 0 { + limit = int(request.Limit) - len(response.Chunk) + if limit <= 0 { + return util.JSONResponse{ + Code: http.StatusOK, + JSON: response, + } + } + } + + // downcasting `limit` is safe as we know it isn't bigger than request.Limit which is int16 + fedRooms := bulkFetchPublicRoomsFromServers(req.Context(), fedClient, extRoomsProvider.Homeservers(), int16(limit)) + response.Chunk = append(response.Chunk, fedRooms...) + return util.JSONResponse{ + Code: http.StatusOK, + JSON: response, + } +} + +// bulkFetchPublicRoomsFromServers fetches public rooms from the list of homeservers. +// Returns a list of public rooms up to the limit specified. +func bulkFetchPublicRoomsFromServers( + ctx context.Context, fedClient *gomatrixserverlib.FederationClient, homeservers []string, limit int16, +) (publicRooms []gomatrixserverlib.PublicRoom) { + // follow pipeline semantics, see https://blog.golang.org/pipelines for more info. + // goroutines send rooms to this channel + roomCh := make(chan gomatrixserverlib.PublicRoom, int(limit)) + // signalling channel to tell goroutines to stop sending rooms and quit + done := make(chan bool) + // signalling to say when we can close the room channel + var wg sync.WaitGroup + wg.Add(len(homeservers)) + // concurrently query for public rooms + for _, hs := range homeservers { + go func(homeserverDomain string) { + defer wg.Done() + util.GetLogger(ctx).WithField("hs", homeserverDomain).Info("Querying HS for public rooms") + fres, err := fedClient.GetPublicRooms(ctx, gomatrixserverlib.ServerName(homeserverDomain), int(limit), "", false, "") + if err != nil { + util.GetLogger(ctx).WithError(err).WithField("hs", homeserverDomain).Warn( + "bulkFetchPublicRoomsFromServers: failed to query hs", + ) + return + } + for _, room := range fres.Chunk { + // atomically send a room or stop + select { + case roomCh <- room: + case <-done: + util.GetLogger(ctx).WithError(err).WithField("hs", homeserverDomain).Info("Interruped whilst sending rooms") + return + } + } + }(hs) + } + + // Close the room channel when the goroutines have quit so we don't leak, but don't let it stop the in-flight request. + // This also allows the request to fail fast if all HSes experience errors as it will cause the room channel to be + // closed. + go func() { + wg.Wait() + util.GetLogger(ctx).Info("Cleaning up resources") + close(roomCh) + }() + + // fan-in results with timeout. We stop when we reach the limit. +FanIn: + for len(publicRooms) < int(limit) || limit == 0 { + // add a room or timeout + select { + case room, ok := <-roomCh: + if !ok { + util.GetLogger(ctx).Info("All homeservers have been queried, returning results.") + break FanIn + } + publicRooms = append(publicRooms, room) + case <-time.After(15 * time.Second): // we've waited long enough, let's tell the client what we got. + util.GetLogger(ctx).Info("Waited 15s for federated public rooms, returning early") + break FanIn + case <-ctx.Done(): // the client hung up on us, let's stop. + util.GetLogger(ctx).Info("Client hung up, returning early") + break FanIn + } + } + // tell goroutines to stop + close(done) + + return publicRooms +} + +func publicRooms(ctx context.Context, request PublicRoomReq, publicRoomDatabase storage.Database) (*gomatrixserverlib.RespPublicRooms, error) { var response gomatrixserverlib.RespPublicRooms var limit int16 var offset int64 diff --git a/publicroomsapi/publicroomsapi.go b/publicroomsapi/publicroomsapi.go index ff0bdb29b..399c0cc57 100644 --- a/publicroomsapi/publicroomsapi.go +++ b/publicroomsapi/publicroomsapi.go @@ -22,6 +22,7 @@ import ( "github.com/matrix-org/dendrite/publicroomsapi/storage" "github.com/matrix-org/dendrite/publicroomsapi/types" roomserverAPI "github.com/matrix-org/dendrite/roomserver/api" + "github.com/matrix-org/gomatrixserverlib" "github.com/sirupsen/logrus" ) @@ -31,6 +32,7 @@ func SetupPublicRoomsAPIComponent( base *basecomponent.BaseDendrite, deviceDB devices.Database, rsQueryAPI roomserverAPI.RoomserverQueryAPI, + fedClient *gomatrixserverlib.FederationClient, extRoomsProvider types.ExternalPublicRoomsProvider, ) { publicRoomsDB, err := storage.NewPublicRoomsServerDatabase(string(base.Cfg.Database.PublicRoomsAPI)) @@ -45,5 +47,5 @@ func SetupPublicRoomsAPIComponent( logrus.WithError(err).Panic("failed to start public rooms server consumer") } - routing.Setup(base.APIMux, deviceDB, publicRoomsDB, extRoomsProvider) + routing.Setup(base.APIMux, deviceDB, publicRoomsDB, fedClient, extRoomsProvider) } diff --git a/publicroomsapi/routing/routing.go b/publicroomsapi/routing/routing.go index 32365e4fc..321b61b89 100644 --- a/publicroomsapi/routing/routing.go +++ b/publicroomsapi/routing/routing.go @@ -15,9 +15,7 @@ package routing import ( - "fmt" "net/http" - "time" "github.com/gorilla/mux" "github.com/matrix-org/dendrite/clientapi/auth" @@ -27,6 +25,7 @@ import ( "github.com/matrix-org/dendrite/publicroomsapi/directory" "github.com/matrix-org/dendrite/publicroomsapi/storage" "github.com/matrix-org/dendrite/publicroomsapi/types" + "github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/util" ) @@ -37,7 +36,10 @@ const pathPrefixR0 = "/_matrix/client/r0" // Due to Setup being used to call many other functions, a gocyclo nolint is // applied: // nolint: gocyclo -func Setup(apiMux *mux.Router, deviceDB devices.Database, publicRoomsDB storage.Database, extRoomsProvider types.ExternalPublicRoomsProvider) { +func Setup( + apiMux *mux.Router, deviceDB devices.Database, publicRoomsDB storage.Database, + fedClient *gomatrixserverlib.FederationClient, extRoomsProvider types.ExternalPublicRoomsProvider, +) { r0mux := apiMux.PathPrefix(pathPrefixR0).Subrouter() authData := auth.Data{ @@ -68,6 +70,7 @@ func Setup(apiMux *mux.Router, deviceDB devices.Database, publicRoomsDB storage. r0mux.Handle("/publicRooms", common.MakeExternalAPI("public_rooms", func(req *http.Request) util.JSONResponse { if extRoomsProvider != nil { + return directory.GetPostPublicRoomsWithExternal(req, publicRoomsDB, fedClient, extRoomsProvider) } return directory.GetPostPublicRooms(req, publicRoomsDB) }), @@ -79,18 +82,4 @@ func Setup(apiMux *mux.Router, deviceDB devices.Database, publicRoomsDB storage. return directory.GetPostPublicRooms(req, publicRoomsDB) }), ).Methods(http.MethodGet) - - if extRoomsProvider != nil { - pollExternalPublicRooms(extRoomsProvider) - } -} - -func pollExternalPublicRooms(extRoomsProvider types.ExternalPublicRoomsProvider) { - go func() { - for { - hses := extRoomsProvider.Homeservers() - fmt.Println(hses) - time.Sleep(extRoomsProvider.PollInterval()) - } - }() } diff --git a/publicroomsapi/storage/sqlite3/public_rooms_table.go b/publicroomsapi/storage/sqlite3/public_rooms_table.go index b8d59292a..44679837f 100644 --- a/publicroomsapi/storage/sqlite3/public_rooms_table.go +++ b/publicroomsapi/storage/sqlite3/public_rooms_table.go @@ -22,6 +22,7 @@ import ( "errors" "fmt" + "github.com/matrix-org/dendrite/common" "github.com/matrix-org/gomatrixserverlib" ) @@ -66,7 +67,7 @@ const selectPublicRoomsWithLimitSQL = "" + "SELECT room_id, joined_members, aliases, canonical_alias, name, topic, world_readable, guest_can_join, avatar_url" + " FROM publicroomsapi_public_rooms WHERE visibility = true" + " ORDER BY joined_members DESC" + - " LIMIT $2 OFFSET $1" + " LIMIT $1 OFFSET $2" const selectPublicRoomsWithFilterSQL = "" + "SELECT room_id, joined_members, aliases, canonical_alias, name, topic, world_readable, guest_can_join, avatar_url" + @@ -192,6 +193,7 @@ func (s *publicRoomsStatements) selectPublicRooms( if err != nil { return []gomatrixserverlib.PublicRoom{}, nil } + defer common.CloseAndLogIfError(ctx, rows, "selectPublicRooms failed to close rows") rooms := []gomatrixserverlib.PublicRoom{} for rows.Next() { diff --git a/publicroomsapi/types/types.go b/publicroomsapi/types/types.go index 252645699..11cb0d204 100644 --- a/publicroomsapi/types/types.go +++ b/publicroomsapi/types/types.go @@ -14,14 +14,11 @@ package types -import "time" - // ExternalPublicRoomsProvider provides a list of homeservers who should be queried // periodically for a list of public rooms on their server. type ExternalPublicRoomsProvider interface { - // The interval at which to check servers - PollInterval() time.Duration // The list of homeserver domains to query. These servers will receive a request // via this API: https://matrix.org/docs/spec/server_server/latest#public-room-directory + // This will be called -on demand- by clients, so cache appropriately! Homeservers() []string }