mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2026-01-17 02:53:11 -06:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into mailbox

Browse source
This commit is contained in:
Devon Hudson 2022-12-19 09:53:10 -07:00
parent 30196ef083 76db8e90de
commit 9f8a78774a
No known key found for this signature in database
GPG key ID: CD06B18E77F6A628
79 changed files with 1455 additions and 394 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
.github/codecov.yaml vendored Normal file
View file

@ -0,0 +1,20 @@
flag_management:
default_rules:
carryforward: true
coverage:
status:
project:
default:
target: auto
threshold: 0%
base: auto
flags:
- unittests
patch:
default:
target: 75%
threshold: 0%
base: auto
flags:
- unittests

3
.github/workflows/dendrite.yml vendored
View file

@ -331,7 +331,8 @@ jobs:
postgres: postgres
api: full-http
container:
image: matrixdotorg/sytest-dendrite:latest
# Temporary for debugging to see if this image is working better.
image: matrixdotorg/sytest-dendrite@sha256:434ad464a9f4ed3f8c3cc47200275b6ccb5c5031a8063daf4acea62be5a23c73
volumes:
- ${{ github.workspace }}:/src
- /root/.cache/go-build:/github/home/.cache/go-build

7
appservice/appservice.go
View file

@ -24,6 +24,8 @@ import (
"github.com/gorilla/mux"
"github.com/sirupsen/logrus"
"github.com/matrix-org/gomatrixserverlib"
appserviceAPI "github.com/matrix-org/dendrite/appservice/api"
"github.com/matrix-org/dendrite/appservice/consumers"
"github.com/matrix-org/dendrite/appservice/inthttp"
@ -32,12 +34,11 @@ import (
"github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
userapi "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/gomatrixserverlib"
)
// AddInternalRoutes registers HTTP handlers for internal API calls
func AddInternalRoutes(router *mux.Router, queryAPI appserviceAPI.AppServiceInternalAPI) {
inthttp.AddRoutes(queryAPI, router)
func AddInternalRoutes(router *mux.Router, queryAPI appserviceAPI.AppServiceInternalAPI, enableMetrics bool) {
inthttp.AddRoutes(queryAPI, router, enableMetrics)
}
// NewInternalAPI returns a concerete implementation of the internal API. Callers

79
appservice/appservice_test.go
View file

@ -77,32 +77,6 @@ func TestAppserviceInternalAPI(t *testing.T) {
}
}))
// TODO: use test.WithAllDatabases
// only one DBType, since appservice.AddInternalRoutes complains about multiple prometheus counters added
base, closeBase := testrig.CreateBaseDendrite(t, test.DBTypeSQLite)
defer closeBase()
// Create a dummy application service
base.Cfg.AppServiceAPI.Derived.ApplicationServices = []config.ApplicationService{
{
ID: "someID",
URL: srv.URL,
ASToken: "",
HSToken: "",
SenderLocalpart: "senderLocalPart",
NamespaceMap: map[string][]config.ApplicationServiceNamespace{
"users": {{RegexpObject: regexp.MustCompile("as-.*")}},
"aliases": {{RegexpObject: regexp.MustCompile("asroom-.*")}},
},
Protocols: []string{existingProtocol},
},
}
// Create required internal APIs
rsAPI := roomserver.NewInternalAPI(base)
usrAPI := userapi.NewInternalAPI(base, &base.Cfg.UserAPI, nil, nil, rsAPI, nil)
asAPI := appservice.NewInternalAPI(base, usrAPI, rsAPI)
// The test cases to run
runCases := func(t *testing.T, testAPI api.AppServiceInternalAPI) {
t.Run("UserIDExists", func(t *testing.T) {
@ -133,24 +107,49 @@ func TestAppserviceInternalAPI(t *testing.T) {
})
}
// Finally execute the tests
t.Run("HTTP API", func(t *testing.T) {
router := mux.NewRouter().PathPrefix(httputil.InternalPathPrefix).Subrouter()
appservice.AddInternalRoutes(router, asAPI)
apiURL, cancel := test.ListenAndServe(t, router, false)
defer cancel()
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
base, closeBase := testrig.CreateBaseDendrite(t, dbType)
defer closeBase()
asHTTPApi, err := inthttp.NewAppserviceClient(apiURL, &http.Client{})
if err != nil {
t.Fatalf("failed to create HTTP client: %s", err)
// Create a dummy application service
base.Cfg.AppServiceAPI.Derived.ApplicationServices = []config.ApplicationService{
{
ID: "someID",
URL: srv.URL,
ASToken: "",
HSToken: "",
SenderLocalpart: "senderLocalPart",
NamespaceMap: map[string][]config.ApplicationServiceNamespace{
"users": {{RegexpObject: regexp.MustCompile("as-.*")}},
"aliases": {{RegexpObject: regexp.MustCompile("asroom-.*")}},
},
Protocols: []string{existingProtocol},
},
}
runCases(t, asHTTPApi)
})
t.Run("Monolith", func(t *testing.T) {
runCases(t, asAPI)
})
// Create required internal APIs
rsAPI := roomserver.NewInternalAPI(base)
usrAPI := userapi.NewInternalAPI(base, &base.Cfg.UserAPI, nil, nil, rsAPI, nil)
asAPI := appservice.NewInternalAPI(base, usrAPI, rsAPI)
// Finally execute the tests
t.Run("HTTP API", func(t *testing.T) {
router := mux.NewRouter().PathPrefix(httputil.InternalPathPrefix).Subrouter()
appservice.AddInternalRoutes(router, asAPI, base.EnableMetrics)
apiURL, cancel := test.ListenAndServe(t, router, false)
defer cancel()
asHTTPApi, err := inthttp.NewAppserviceClient(apiURL, &http.Client{})
if err != nil {
t.Fatalf("failed to create HTTP client: %s", err)
}
runCases(t, asHTTPApi)
})
t.Run("Monolith", func(t *testing.T) {
runCases(t, asAPI)
})
})
}
func testUserIDExists(t *testing.T, asAPI api.AppServiceInternalAPI, userID string, wantExists bool) {

12
appservice/inthttp/server.go
View file

@ -8,29 +8,29 @@ import (
)
// AddRoutes adds the AppServiceQueryAPI handlers to the http.ServeMux.
func AddRoutes(a api.AppServiceInternalAPI, internalAPIMux *mux.Router) {
func AddRoutes(a api.AppServiceInternalAPI, internalAPIMux *mux.Router, enableMetrics bool) {
internalAPIMux.Handle(
AppServiceRoomAliasExistsPath,
httputil.MakeInternalRPCAPI("AppserviceRoomAliasExists", a.RoomAliasExists),
httputil.MakeInternalRPCAPI("AppserviceRoomAliasExists", enableMetrics, a.RoomAliasExists),
)
internalAPIMux.Handle(
AppServiceUserIDExistsPath,
httputil.MakeInternalRPCAPI("AppserviceUserIDExists", a.UserIDExists),
httputil.MakeInternalRPCAPI("AppserviceUserIDExists", enableMetrics, a.UserIDExists),
)
internalAPIMux.Handle(
AppServiceProtocolsPath,
httputil.MakeInternalRPCAPI("AppserviceProtocols", a.Protocols),
httputil.MakeInternalRPCAPI("AppserviceProtocols", enableMetrics, a.Protocols),
)
internalAPIMux.Handle(
AppServiceLocationsPath,
httputil.MakeInternalRPCAPI("AppserviceLocations", a.Locations),
httputil.MakeInternalRPCAPI("AppserviceLocations", enableMetrics, a.Locations),
)
internalAPIMux.Handle(
AppServiceUserPath,
httputil.MakeInternalRPCAPI("AppserviceUser", a.User),
httputil.MakeInternalRPCAPI("AppserviceUser", enableMetrics, a.User),
)
}

6
build/dendritejs-pinecone/main.go
View file

@ -180,14 +180,14 @@ func startup() {
base := base.NewBaseDendrite(cfg, "Monolith")
defer base.Close() // nolint: errcheck
rsAPI := roomserver.NewInternalAPI(base)
federation := conn.CreateFederationClient(base, pSessions)
keyAPI := keyserver.NewInternalAPI(base, &base.Cfg.KeyServer, federation)
keyAPI := keyserver.NewInternalAPI(base, &base.Cfg.KeyServer, federation, rsAPI)
serverKeyAPI := &signing.YggdrasilKeys{}
keyRing := serverKeyAPI.KeyRing()
rsAPI := roomserver.NewInternalAPI(base)
userAPI := userapi.NewInternalAPI(base, &cfg.UserAPI, nil, keyAPI, rsAPI, base.PushGatewayHTTPClient())
keyAPI.SetUserAPI(userAPI)

2
build/gobind-pinecone/monolith.go
View file

@ -353,7 +353,7 @@ func (m *DendriteMonolith) Start() {
base, federation, rsAPI, base.Caches, keyRing, true,
)
keyAPI := keyserver.NewInternalAPI(base, &base.Cfg.KeyServer, m.federationAPI)
keyAPI := keyserver.NewInternalAPI(base, &base.Cfg.KeyServer, m.federationAPI, rsAPI)
m.userAPI = userapi.NewInternalAPI(base, &cfg.UserAPI, cfg.Derived.ApplicationServices, keyAPI, rsAPI, base.PushGatewayHTTPClient())
keyAPI.SetUserAPI(m.userAPI)

2
build/gobind-yggdrasil/monolith.go
View file

@ -165,7 +165,7 @@ func (m *DendriteMonolith) Start() {
base, federation, rsAPI, base.Caches, keyRing, true,
)
keyAPI := keyserver.NewInternalAPI(base, &base.Cfg.KeyServer, federation)
keyAPI := keyserver.NewInternalAPI(base, &base.Cfg.KeyServer, federation, rsAPI)
userAPI := userapi.NewInternalAPI(base, &cfg.UserAPI, cfg.Derived.ApplicationServices, keyAPI, rsAPI, base.PushGatewayHTTPClient())
keyAPI.SetUserAPI(userAPI)

7
cmd/dendrite-demo-pinecone/main.go
View file

@ -221,12 +221,7 @@ func main() {
base, federation, rsAPI, base.Caches, keyRing, true,
)
// TODO : What if I had another dendrite function, where I just passed in the fed_client,
// rsAPI, userAPI, keyRing & any other info. Then that thing is what handles doing the
// async_events querying?
// ie. don't tie it into the federationInternalAPI & get rid of the associated mess.
keyAPI := keyserver.NewInternalAPI(base, &base.Cfg.KeyServer, fsAPI)
keyAPI := keyserver.NewInternalAPI(base, &base.Cfg.KeyServer, fsAPI, rsComponent)
userAPI := userapi.NewInternalAPI(base, &cfg.UserAPI, nil, keyAPI, rsAPI, base.PushGatewayHTTPClient())
keyAPI.SetUserAPI(userAPI)

5
cmd/dendrite-demo-yggdrasil/main.go
View file

@ -157,11 +157,12 @@ func main() {
serverKeyAPI := &signing.YggdrasilKeys{}
keyRing := serverKeyAPI.KeyRing()
keyAPI := keyserver.NewInternalAPI(base, &base.Cfg.KeyServer, federation)
rsComponent := roomserver.NewInternalAPI(
base,
)
keyAPI := keyserver.NewInternalAPI(base, &base.Cfg.KeyServer, federation, rsComponent)
rsAPI := rsComponent
userAPI := userapi.NewInternalAPI(base, &cfg.UserAPI, nil, keyAPI, rsAPI, base.PushGatewayHTTPClient())

15
cmd/dendrite-monolith-server/main.go
View file

@ -18,6 +18,8 @@ import (
"flag"
"os"
"github.com/sirupsen/logrus"
"github.com/matrix-org/dendrite/appservice"
"github.com/matrix-org/dendrite/federationapi"
"github.com/matrix-org/dendrite/keyserver"
@ -29,7 +31,6 @@ import (
"github.com/matrix-org/dendrite/setup/mscs"
"github.com/matrix-org/dendrite/userapi"
uapi "github.com/matrix-org/dendrite/userapi/api"
"github.com/sirupsen/logrus"
)
var (
@ -75,7 +76,7 @@ func main() {
// call functions directly on the impl unless running in HTTP mode
rsAPI := rsImpl
if base.UseHTTPAPIs {
roomserver.AddInternalRoutes(base.InternalAPIMux, rsImpl)
roomserver.AddInternalRoutes(base.InternalAPIMux, rsImpl, base.EnableMetrics)
rsAPI = base.RoomserverHTTPClient()
}
if traceInternal {
@ -89,15 +90,15 @@ func main() {
)
fsImplAPI := fsAPI
if base.UseHTTPAPIs {
federationapi.AddInternalRoutes(base.InternalAPIMux, fsAPI)
federationapi.AddInternalRoutes(base.InternalAPIMux, fsAPI, base.EnableMetrics)
fsAPI = base.FederationAPIHTTPClient()
}
keyRing := fsAPI.KeyRing()
keyImpl := keyserver.NewInternalAPI(base, &base.Cfg.KeyServer, fsAPI)
keyImpl := keyserver.NewInternalAPI(base, &base.Cfg.KeyServer, fsAPI, rsAPI)
keyAPI := keyImpl
if base.UseHTTPAPIs {
keyserver.AddInternalRoutes(base.InternalAPIMux, keyAPI)
keyserver.AddInternalRoutes(base.InternalAPIMux, keyAPI, base.EnableMetrics)
keyAPI = base.KeyServerHTTPClient()
}
@ -105,7 +106,7 @@ func main() {
userImpl := userapi.NewInternalAPI(base, &cfg.UserAPI, cfg.Derived.ApplicationServices, keyAPI, rsAPI, pgClient)
userAPI := userImpl
if base.UseHTTPAPIs {
userapi.AddInternalRoutes(base.InternalAPIMux, userAPI)
userapi.AddInternalRoutes(base.InternalAPIMux, userAPI, base.EnableMetrics)
userAPI = base.UserAPIClient()
}
if traceInternal {
@ -119,7 +120,7 @@ func main() {
// before the listeners are up.
asAPI := appservice.NewInternalAPI(base, userImpl, rsAPI)
if base.UseHTTPAPIs {
appservice.AddInternalRoutes(base.InternalAPIMux, asAPI)
appservice.AddInternalRoutes(base.InternalAPIMux, asAPI, base.EnableMetrics)
asAPI = base.AppserviceHTTPClient()
}

2
cmd/dendrite-polylith-multi/personalities/appservice.go
View file

@ -26,7 +26,7 @@ func Appservice(base *base.BaseDendrite, cfg *config.Dendrite) {
rsAPI := base.RoomserverHTTPClient()
intAPI := appservice.NewInternalAPI(base, userAPI, rsAPI)
appservice.AddInternalRoutes(base.InternalAPIMux, intAPI)
appservice.AddInternalRoutes(base.InternalAPIMux, intAPI, base.EnableMetrics)
base.SetupAndServeHTTP(
base.Cfg.AppServiceAPI.InternalAPI.Listen, // internal listener

2
cmd/dendrite-polylith-multi/personalities/federationapi.go
View file

@ -34,7 +34,7 @@ func FederationAPI(base *basepkg.BaseDendrite, cfg *config.Dendrite) {
rsAPI, fsAPI, keyAPI, nil,
)
federationapi.AddInternalRoutes(base.InternalAPIMux, fsAPI)
federationapi.AddInternalRoutes(base.InternalAPIMux, fsAPI, base.EnableMetrics)
base.SetupAndServeHTTP(
base.Cfg.FederationAPI.InternalAPI.Listen,

5
cmd/dendrite-polylith-multi/personalities/keyserver.go
View file

@ -22,10 +22,11 @@ import (
func KeyServer(base *basepkg.BaseDendrite, cfg *config.Dendrite) {
fsAPI := base.FederationAPIHTTPClient()
intAPI := keyserver.NewInternalAPI(base, &base.Cfg.KeyServer, fsAPI)
rsAPI := base.RoomserverHTTPClient()
intAPI := keyserver.NewInternalAPI(base, &base.Cfg.KeyServer, fsAPI, rsAPI)
intAPI.SetUserAPI(base.UserAPIClient())
keyserver.AddInternalRoutes(base.InternalAPIMux, intAPI)
keyserver.AddInternalRoutes(base.InternalAPIMux, intAPI, base.EnableMetrics)
base.SetupAndServeHTTP(
base.Cfg.KeyServer.InternalAPI.Listen, // internal listener

2
cmd/dendrite-polylith-multi/personalities/roomserver.go
View file

@ -26,7 +26,7 @@ func RoomServer(base *basepkg.BaseDendrite, cfg *config.Dendrite) {
rsAPI := roomserver.NewInternalAPI(base)
rsAPI.SetFederationAPI(fsAPI, fsAPI.KeyRing())
rsAPI.SetAppserviceAPI(asAPI)
roomserver.AddInternalRoutes(base.InternalAPIMux, rsAPI)
roomserver.AddInternalRoutes(base.InternalAPIMux, rsAPI, base.EnableMetrics)
base.SetupAndServeHTTP(
base.Cfg.RoomServer.InternalAPI.Listen, // internal listener

2
cmd/dendrite-polylith-multi/personalities/userapi.go
View file

@ -27,7 +27,7 @@ func UserAPI(base *basepkg.BaseDendrite, cfg *config.Dendrite) {
base.PushGatewayHTTPClient(),
)
userapi.AddInternalRoutes(base.InternalAPIMux, userAPI)
userapi.AddInternalRoutes(base.InternalAPIMux, userAPI, base.EnableMetrics)
base.SetupAndServeHTTP(
base.Cfg.UserAPI.InternalAPI.Listen, // internal listener

4
docs/FAQ.md
View file

@ -91,7 +91,7 @@ Please use PostgreSQL wherever possible, especially if you are planning to run a
## Dendrite is using a lot of CPU
Generally speaking, you should expect to see some CPU spikes, particularly if you are joining or participating in large rooms. However, constant/sustained high CPU usage is not expected - if you are experiencing that, please join `#dendrite-dev:matrix.org` and let us know what you were doing when the
CPU usage shot up, or file a GitHub issue. If you can take a [CPU profile](PROFILING.md) then that would
CPU usage shot up, or file a GitHub issue. If you can take a [CPU profile](development/PROFILING.md) then that would
be a huge help too, as that will help us to understand where the CPU time is going.
## Dendrite is using a lot of RAM
@ -99,7 +99,7 @@ be a huge help too, as that will help us to understand where the CPU time is goi
As above with CPU usage, some memory spikes are expected if Dendrite is doing particularly heavy work
at a given instant. However, if it is using more RAM than you expect for a long time, that's probably
not expected. Join `#dendrite-dev:matrix.org` and let us know what you were doing when the memory usage
ballooned, or file a GitHub issue if you can. If you can take a [memory profile](PROFILING.md) then that
ballooned, or file a GitHub issue if you can. If you can take a [memory profile](development/PROFILING.md) then that
would be a huge help too, as that will help us to understand where the memory usage is happening.
## Dendrite is running out of PostgreSQL database connections

6
docs/Gemfile.lock
View file

@ -231,9 +231,9 @@ GEM
jekyll-seo-tag (~> 2.1)
minitest (5.15.0)
multipart-post (2.1.1)
nokogiri (1.13.9-arm64-darwin)
nokogiri (1.13.10-arm64-darwin)
racc (~> 1.4)
nokogiri (1.13.9-x86_64-linux)
nokogiri (1.13.10-x86_64-linux)
racc (~> 1.4)
octokit (4.22.0)
faraday (>= 0.9)
@ -241,7 +241,7 @@ GEM
pathutil (0.16.2)
forwardable-extended (~> 2.6)
public_suffix (4.0.7)
racc (1.6.0)
racc (1.6.1)
rb-fsevent (0.11.1)
rb-inotify (0.10.1)
ffi (~> 1.0)

26
docs/CONTRIBUTING.md → docs/development/CONTRIBUTING.md
View file

@ -9,6 +9,28 @@ permalink: /development/contributing
Everyone is welcome to contribute to Dendrite! We aim to make it as easy as
possible to get started.
## Contribution types
We are a small team maintaining a large project. As a result, we cannot merge every feature, even if it
is bug-free and useful, because we then commit to maintaining it indefinitely. We will always accept:
- bug fixes
- security fixes (please responsibly disclose via security@matrix.org *before* creating pull requests)
We will accept the following with caveats:
- documentation fixes, provided they do not add additional instructions which can end up going out-of-date,
e.g example configs, shell commands.
- performance fixes, provided they do not add significantly more maintenance burden.
- additional functionality on existing features, provided the functionality is small and maintainable.
- additional functionality that, in its absence, would impact the ecosystem e.g spam and abuse mitigations
- test-only changes, provided they help improve coverage or test tricky code.
The following items are at risk of not being accepted:
- Configuration or CLI changes, particularly ones which increase the overall configuration surface.
The following items are unlikely to be accepted into a main Dendrite release for now:
- New MSC implementations.
- New features which are not in the specification.
## Sign off
We require that everyone who contributes to the project signs off their contributions
@ -35,7 +57,7 @@ to do so for future contributions.
## Getting up and running
See the [Installation](installation) section for information on how to build an
See the [Installation](../installation) section for information on how to build an
instance of Dendrite. You will likely need this in order to test your changes.
## Code style
@ -129,7 +151,7 @@ significant amount of CPU and RAM.
Once the code builds, run [Sytest](https://github.com/matrix-org/sytest)
according to the guide in
[docs/sytest.md](https://github.com/matrix-org/dendrite/blob/main/docs/sytest.md#using-a-sytest-docker-image)
[docs/development/sytest.md](https://github.com/matrix-org/dendrite/blob/main/docs/development/sytest.md#using-a-sytest-docker-image)
so you can see whether something is being broken and whether there are newly
passing tests.

0
docs/PROFILING.md → docs/development/PROFILING.md
View file

0
docs/coverage.md → docs/development/coverage.md
View file

0
docs/sytest.md → docs/development/sytest.md
View file

0
docs/tracing/opentracing.md → docs/development/tracing/opentracing.md
View file

0
docs/tracing/setup.md → docs/development/tracing/setup.md
View file

2
federationapi/consumers/roomserver.go
View file

@ -232,7 +232,7 @@ func (s *OutputRoomEventConsumer) processMessage(ore api.OutputNewRoomEvent, rew
}
func (s *OutputRoomEventConsumer) sendPresence(roomID string, addedJoined []types.JoinedHost) {
joined := make([]gomatrixserverlib.ServerName, len(addedJoined))
joined := make([]gomatrixserverlib.ServerName, 0, len(addedJoined))
for _, added := range addedJoined {
joined = append(joined, added.ServerName)
}

4
federationapi/federationapi.go
View file

@ -43,8 +43,8 @@ import (
// AddInternalRoutes registers HTTP handlers for the internal API. Invokes functions
// on the given input API.
func AddInternalRoutes(router *mux.Router, intAPI api.FederationInternalAPI) {
inthttp.AddRoutes(intAPI, router)
func AddInternalRoutes(router *mux.Router, intAPI api.FederationInternalAPI, enableMetrics bool) {
inthttp.AddRoutes(intAPI, router, enableMetrics)
}
// AddPublicRoutes sets up and registers HTTP handlers on the base API muxes for the FederationAPI component.

46
federationapi/inthttp/server.go
View file

@ -17,41 +17,41 @@ import (
// AddRoutes adds the FederationInternalAPI handlers to the http.ServeMux.
// nolint:gocyclo
func AddRoutes(intAPI api.FederationInternalAPI, internalAPIMux *mux.Router) {
func AddRoutes(intAPI api.FederationInternalAPI, internalAPIMux *mux.Router, enableMetrics bool) {
internalAPIMux.Handle(
FederationAPIQueryJoinedHostServerNamesInRoomPath,
httputil.MakeInternalRPCAPI("FederationAPIQueryJoinedHostServerNamesInRoom", intAPI.QueryJoinedHostServerNamesInRoom),
httputil.MakeInternalRPCAPI("FederationAPIQueryJoinedHostServerNamesInRoom", enableMetrics, intAPI.QueryJoinedHostServerNamesInRoom),
)
internalAPIMux.Handle(
FederationAPIPerformInviteRequestPath,
httputil.MakeInternalRPCAPI("FederationAPIPerformInvite", intAPI.PerformInvite),
httputil.MakeInternalRPCAPI("FederationAPIPerformInvite", enableMetrics, intAPI.PerformInvite),
)
internalAPIMux.Handle(
FederationAPIPerformLeaveRequestPath,
httputil.MakeInternalRPCAPI("FederationAPIPerformLeave", intAPI.PerformLeave),
httputil.MakeInternalRPCAPI("FederationAPIPerformLeave", enableMetrics, intAPI.PerformLeave),
)
internalAPIMux.Handle(
FederationAPIPerformDirectoryLookupRequestPath,
httputil.MakeInternalRPCAPI("FederationAPIPerformDirectoryLookupRequest", intAPI.PerformDirectoryLookup),
httputil.MakeInternalRPCAPI("FederationAPIPerformDirectoryLookupRequest", enableMetrics, intAPI.PerformDirectoryLookup),
)
internalAPIMux.Handle(
FederationAPIPerformBroadcastEDUPath,
httputil.MakeInternalRPCAPI("FederationAPIPerformBroadcastEDU", intAPI.PerformBroadcastEDU),
httputil.MakeInternalRPCAPI("FederationAPIPerformBroadcastEDU", enableMetrics, intAPI.PerformBroadcastEDU),
)
internalAPIMux.Handle(
FederationAPIPerformWakeupServers,
httputil.MakeInternalRPCAPI("FederationAPIPerformWakeupServers", intAPI.PerformWakeupServers),
httputil.MakeInternalRPCAPI("FederationAPIPerformWakeupServers", enableMetrics, intAPI.PerformWakeupServers),
)
internalAPIMux.Handle(
FederationAPIPerformJoinRequestPath,
httputil.MakeInternalRPCAPI(
"FederationAPIPerformJoinRequest",
"FederationAPIPerformJoinRequest", enableMetrics,
func(ctx context.Context, req *api.PerformJoinRequest, res *api.PerformJoinResponse) error {
intAPI.PerformJoin(ctx, req, res)
return nil
@ -62,7 +62,7 @@ func AddRoutes(intAPI api.FederationInternalAPI, internalAPIMux *mux.Router) {
internalAPIMux.Handle(
FederationAPIGetUserDevicesPath,
httputil.MakeInternalProxyAPI(
"FederationAPIGetUserDevices",
"FederationAPIGetUserDevices", enableMetrics,
func(ctx context.Context, req *getUserDevices) (*gomatrixserverlib.RespUserDevices, error) {
res, err := intAPI.GetUserDevices(ctx, req.Origin, req.S, req.UserID)
return &res, federationClientError(err)
@ -73,7 +73,7 @@ func AddRoutes(intAPI api.FederationInternalAPI, internalAPIMux *mux.Router) {
internalAPIMux.Handle(
FederationAPIClaimKeysPath,
httputil.MakeInternalProxyAPI(
"FederationAPIClaimKeys",
"FederationAPIClaimKeys", enableMetrics,
func(ctx context.Context, req *claimKeys) (*gomatrixserverlib.RespClaimKeys, error) {
res, err := intAPI.ClaimKeys(ctx, req.Origin, req.S, req.OneTimeKeys)
return &res, federationClientError(err)
@ -84,7 +84,7 @@ func AddRoutes(intAPI api.FederationInternalAPI, internalAPIMux *mux.Router) {
internalAPIMux.Handle(
FederationAPIQueryKeysPath,
httputil.MakeInternalProxyAPI(
"FederationAPIQueryKeys",
"FederationAPIQueryKeys", enableMetrics,
func(ctx context.Context, req *queryKeys) (*gomatrixserverlib.RespQueryKeys, error) {
res, err := intAPI.QueryKeys(ctx, req.Origin, req.S, req.Keys)
return &res, federationClientError(err)
@ -95,7 +95,7 @@ func AddRoutes(intAPI api.FederationInternalAPI, internalAPIMux *mux.Router) {
internalAPIMux.Handle(
FederationAPIBackfillPath,
httputil.MakeInternalProxyAPI(
"FederationAPIBackfill",
"FederationAPIBackfill", enableMetrics,
func(ctx context.Context, req *backfill) (*gomatrixserverlib.Transaction, error) {
res, err := intAPI.Backfill(ctx, req.Origin, req.S, req.RoomID, req.Limit, req.EventIDs)
return &res, federationClientError(err)
@ -106,7 +106,7 @@ func AddRoutes(intAPI api.FederationInternalAPI, internalAPIMux *mux.Router) {
internalAPIMux.Handle(
FederationAPILookupStatePath,
httputil.MakeInternalProxyAPI(
"FederationAPILookupState",
"FederationAPILookupState", enableMetrics,
func(ctx context.Context, req *lookupState) (*gomatrixserverlib.RespState, error) {
res, err := intAPI.LookupState(ctx, req.Origin, req.S, req.RoomID, req.EventID, req.RoomVersion)
return &res, federationClientError(err)
@ -117,7 +117,7 @@ func AddRoutes(intAPI api.FederationInternalAPI, internalAPIMux *mux.Router) {
internalAPIMux.Handle(
FederationAPILookupStateIDsPath,
httputil.MakeInternalProxyAPI(
"FederationAPILookupStateIDs",
"FederationAPILookupStateIDs", enableMetrics,
func(ctx context.Context, req *lookupStateIDs) (*gomatrixserverlib.RespStateIDs, error) {
res, err := intAPI.LookupStateIDs(ctx, req.Origin, req.S, req.RoomID, req.EventID)
return &res, federationClientError(err)
@ -128,7 +128,7 @@ func AddRoutes(intAPI api.FederationInternalAPI, internalAPIMux *mux.Router) {
internalAPIMux.Handle(
FederationAPILookupMissingEventsPath,
httputil.MakeInternalProxyAPI(
"FederationAPILookupMissingEvents",
"FederationAPILookupMissingEvents", enableMetrics,
func(ctx context.Context, req *lookupMissingEvents) (*gomatrixserverlib.RespMissingEvents, error) {
res, err := intAPI.LookupMissingEvents(ctx, req.Origin, req.S, req.RoomID, req.Missing, req.RoomVersion)
return &res, federationClientError(err)
@ -139,7 +139,7 @@ func AddRoutes(intAPI api.FederationInternalAPI, internalAPIMux *mux.Router) {
internalAPIMux.Handle(
FederationAPIGetEventPath,
httputil.MakeInternalProxyAPI(
"FederationAPIGetEvent",
"FederationAPIGetEvent", enableMetrics,
func(ctx context.Context, req *getEvent) (*gomatrixserverlib.Transaction, error) {
res, err := intAPI.GetEvent(ctx, req.Origin, req.S, req.EventID)
return &res, federationClientError(err)
@ -150,7 +150,7 @@ func AddRoutes(intAPI api.FederationInternalAPI, internalAPIMux *mux.Router) {
internalAPIMux.Handle(
FederationAPIGetEventAuthPath,
httputil.MakeInternalProxyAPI(
"FederationAPIGetEventAuth",
"FederationAPIGetEventAuth", enableMetrics,
func(ctx context.Context, req *getEventAuth) (*gomatrixserverlib.RespEventAuth, error) {
res, err := intAPI.GetEventAuth(ctx, req.Origin, req.S, req.RoomVersion, req.RoomID, req.EventID)
return &res, federationClientError(err)
@ -160,13 +160,13 @@ func AddRoutes(intAPI api.FederationInternalAPI, internalAPIMux *mux.Router) {
internalAPIMux.Handle(
FederationAPIQueryServerKeysPath,
httputil.MakeInternalRPCAPI("FederationAPIQueryServerKeys", intAPI.QueryServerKeys),
httputil.MakeInternalRPCAPI("FederationAPIQueryServerKeys", enableMetrics, intAPI.QueryServerKeys),
)
internalAPIMux.Handle(
FederationAPILookupServerKeysPath,
httputil.MakeInternalProxyAPI(
"FederationAPILookupServerKeys",
"FederationAPILookupServerKeys", enableMetrics,
func(ctx context.Context, req *lookupServerKeys) (*[]gomatrixserverlib.ServerKeys, error) {
res, err := intAPI.LookupServerKeys(ctx, req.S, req.KeyRequests)
return &res, federationClientError(err)
@ -177,7 +177,7 @@ func AddRoutes(intAPI api.FederationInternalAPI, internalAPIMux *mux.Router) {
internalAPIMux.Handle(
FederationAPIEventRelationshipsPath,
httputil.MakeInternalProxyAPI(
"FederationAPIMSC2836EventRelationships",
"FederationAPIMSC2836EventRelationships", enableMetrics,
func(ctx context.Context, req *eventRelationships) (*gomatrixserverlib.MSC2836EventRelationshipsResponse, error) {
res, err := intAPI.MSC2836EventRelationships(ctx, req.Origin, req.S, req.Req, req.RoomVer)
return &res, federationClientError(err)
@ -188,7 +188,7 @@ func AddRoutes(intAPI api.FederationInternalAPI, internalAPIMux *mux.Router) {
internalAPIMux.Handle(
FederationAPISpacesSummaryPath,
httputil.MakeInternalProxyAPI(
"FederationAPIMSC2946SpacesSummary",
"FederationAPIMSC2946SpacesSummary", enableMetrics,
func(ctx context.Context, req *spacesReq) (*gomatrixserverlib.MSC2946SpacesResponse, error) {
res, err := intAPI.MSC2946Spaces(ctx, req.Origin, req.S, req.RoomID, req.SuggestedOnly)
return &res, federationClientError(err)
@ -198,7 +198,7 @@ func AddRoutes(intAPI api.FederationInternalAPI, internalAPIMux *mux.Router) {
// TODO: Look at this shape
internalAPIMux.Handle(FederationAPIQueryPublicKeyPath,
httputil.MakeInternalAPI("FederationAPIQueryPublicKeys", func(req *http.Request) util.JSONResponse {
httputil.MakeInternalAPI("FederationAPIQueryPublicKeys", enableMetrics, func(req *http.Request) util.JSONResponse {
request := api.QueryPublicKeysRequest{}
response := api.QueryPublicKeysResponse{}
if err := json.NewDecoder(req.Body).Decode(&request); err != nil {
@ -215,7 +215,7 @@ func AddRoutes(intAPI api.FederationInternalAPI, internalAPIMux *mux.Router) {
// TODO: Look at this shape
internalAPIMux.Handle(FederationAPIInputPublicKeyPath,
httputil.MakeInternalAPI("FederationAPIInputPublicKeys", func(req *http.Request) util.JSONResponse {
httputil.MakeInternalAPI("FederationAPIInputPublicKeys", enableMetrics, func(req *http.Request) util.JSONResponse {
request := api.InputPublicKeysRequest{}
response := api.InputPublicKeysResponse{}
if err := json.NewDecoder(req.Body).Decode(&request); err != nil {

15
internal/httputil/httpapi.go
View file

@ -24,16 +24,17 @@ import (
"strings"
"github.com/getsentry/sentry-go"
"github.com/matrix-org/dendrite/clientapi/auth"
"github.com/matrix-org/dendrite/clientapi/jsonerror"
userapi "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/util"
opentracing "github.com/opentracing/opentracing-go"
"github.com/opentracing/opentracing-go"
"github.com/opentracing/opentracing-go/ext"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
"github.com/prometheus/client_golang/prometheus/promhttp"
"github.com/sirupsen/logrus"
"github.com/matrix-org/dendrite/clientapi/auth"
"github.com/matrix-org/dendrite/clientapi/jsonerror"
userapi "github.com/matrix-org/dendrite/userapi/api"
)
// BasicAuth is used for authorization on /metrics handlers
@ -227,7 +228,7 @@ func MakeHTMLAPI(metricsName string, f func(http.ResponseWriter, *http.Request)
// This is used for APIs that are internal to dendrite.
// If we are passed a tracing context in the request headers then we use that
// as the parent of any tracing spans we create.
func MakeInternalAPI(metricsName string, f func(*http.Request) util.JSONResponse) http.Handler {
func MakeInternalAPI(metricsName string, enableMetrics bool, f func(*http.Request) util.JSONResponse) http.Handler {
h := util.MakeJSONAPI(util.NewJSONRequestHandler(f))
withSpan := func(w http.ResponseWriter, req *http.Request) {
carrier := opentracing.HTTPHeadersCarrier(req.Header)
@ -246,6 +247,10 @@ func MakeInternalAPI(metricsName string, f func(*http.Request) util.JSONResponse
h.ServeHTTP(w, req)
}
if !enableMetrics {
return http.HandlerFunc(withSpan)
}
return promhttp.InstrumentHandlerCounter(
promauto.NewCounterVec(
prometheus.CounterOpts{

10
internal/httputil/internalapi.go
View file

@ -22,7 +22,7 @@ import (
"reflect"
"github.com/matrix-org/util"
opentracing "github.com/opentracing/opentracing-go"
"github.com/opentracing/opentracing-go"
)
type InternalAPIError struct {
@ -34,8 +34,8 @@ func (e InternalAPIError) Error() string {
return fmt.Sprintf("internal API returned %q error: %s", e.Type, e.Message)
}
func MakeInternalRPCAPI[reqtype, restype any](metricsName string, f func(context.Context, *reqtype, *restype) error) http.Handler {
return MakeInternalAPI(metricsName, func(req *http.Request) util.JSONResponse {
func MakeInternalRPCAPI[reqtype, restype any](metricsName string, enableMetrics bool, f func(context.Context, *reqtype, *restype) error) http.Handler {
return MakeInternalAPI(metricsName, enableMetrics, func(req *http.Request) util.JSONResponse {
var request reqtype
var response restype
if err := json.NewDecoder(req.Body).Decode(&request); err != nil {
@ -57,8 +57,8 @@ func MakeInternalRPCAPI[reqtype, restype any](metricsName string, f func(context
})
}
func MakeInternalProxyAPI[reqtype, restype any](metricsName string, f func(context.Context, *reqtype) (*restype, error)) http.Handler {
return MakeInternalAPI(metricsName, func(req *http.Request) util.JSONResponse {
func MakeInternalProxyAPI[reqtype, restype any](metricsName string, enableMetrics bool, f func(context.Context, *reqtype) (*restype, error)) http.Handler {
return MakeInternalAPI(metricsName, enableMetrics, func(req *http.Request) util.JSONResponse {
var request reqtype
if err := json.NewDecoder(req.Body).Decode(&request); err != nil {
return util.MessageResponse(http.StatusBadRequest, err.Error())

5
internal/log.go
View file

@ -33,6 +33,11 @@ import (
"github.com/matrix-org/dendrite/setup/config"
)
// logrus is using a global variable when we're using `logrus.AddHook`
// this unfortunately results in us adding the same hook multiple times.
// This map ensures we only ever add one level hook.
var stdLevelLogAdded = make(map[logrus.Level]bool)
type utcFormatter struct {
logrus.Formatter
}

13
internal/log_unix.go
View file

@ -22,16 +22,16 @@ import (
"log/syslog"
"github.com/MFAshby/stdemuxerhook"
"github.com/matrix-org/dendrite/setup/config"
"github.com/sirupsen/logrus"
lSyslog "github.com/sirupsen/logrus/hooks/syslog"
"github.com/matrix-org/dendrite/setup/config"
)
// SetupHookLogging configures the logging hooks defined in the configuration.
// If something fails here it means that the logging was improperly configured,
// so we just exit with the error
func SetupHookLogging(hooks []config.LogrusHook, componentName string) {
stdLogAdded := false
for _, hook := range hooks {
// Check we received a proper logging level
level, err := logrus.ParseLevel(hook.Level)
@ -54,14 +54,11 @@ func SetupHookLogging(hooks []config.LogrusHook, componentName string) {
setupSyslogHook(hook, level, componentName)
case "std":
setupStdLogHook(level)
stdLogAdded = true
default:
logrus.Fatalf("Unrecognised logging hook type: %s", hook.Type)
}
}
if !stdLogAdded {
setupStdLogHook(logrus.InfoLevel)
}
setupStdLogHook(logrus.InfoLevel)
// Hooks are now configured for stdout/err, so throw away the default logger output
logrus.SetOutput(io.Discard)
}
@ -88,7 +85,11 @@ func checkSyslogHookParams(params map[string]interface{}) {
}
func setupStdLogHook(level logrus.Level) {
if stdLevelLogAdded[level] {
return
}
logrus.AddHook(&logLevelHook{level, stdemuxerhook.New(logrus.StandardLogger())})
stdLevelLogAdded[level] = true
}
func setupSyslogHook(hook config.LogrusHook, level logrus.Level, componentName string) {

29
keyserver/internal/device_list_update.go
View file

@ -24,6 +24,8 @@ import (
"sync"
"time"
rsapi "github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/gomatrix"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
@ -102,6 +104,7 @@ type DeviceListUpdater struct {
// block on or timeout via a select.
userIDToChan map[string]chan bool
userIDToChanMu *sync.Mutex
rsAPI rsapi.KeyserverRoomserverAPI
}
// DeviceListUpdaterDatabase is the subset of functionality from storage.Database required for the updater.
@ -124,6 +127,8 @@ type DeviceListUpdaterDatabase interface {
// DeviceKeysJSON populates the KeyJSON for the given keys. If any proided `keys` have a `KeyJSON` or `StreamID` already then it will be replaced.
DeviceKeysJSON(ctx context.Context, keys []api.DeviceMessage) error
DeleteStaleDeviceLists(ctx context.Context, userIDs []string) error
}
type DeviceListUpdaterAPI interface {
@ -140,7 +145,7 @@ func NewDeviceListUpdater(
process *process.ProcessContext, db DeviceListUpdaterDatabase,
api DeviceListUpdaterAPI, producer KeyChangeProducer,
fedClient fedsenderapi.KeyserverFederationAPI, numWorkers int,
thisServer gomatrixserverlib.ServerName,
rsAPI rsapi.KeyserverRoomserverAPI, thisServer gomatrixserverlib.ServerName,
) *DeviceListUpdater {
return &DeviceListUpdater{
process: process,
@ -154,6 +159,7 @@ func NewDeviceListUpdater(
workerChans: make([]chan gomatrixserverlib.ServerName, numWorkers),
userIDToChan: make(map[string]chan bool),
userIDToChanMu: &sync.Mutex{},
rsAPI: rsAPI,
}
}
@ -168,7 +174,7 @@ func (u *DeviceListUpdater) Start() error {
go u.worker(ch)
}
staleLists, err := u.db.StaleDeviceLists(context.Background(), []gomatrixserverlib.ServerName{})
staleLists, err := u.db.StaleDeviceLists(u.process.Context(), []gomatrixserverlib.ServerName{})
if err != nil {
return err
}
@ -186,6 +192,25 @@ func (u *DeviceListUpdater) Start() error {
return nil
}
// CleanUp removes stale device entries for users we don't share a room with anymore
func (u *DeviceListUpdater) CleanUp() error {
staleUsers, err := u.db.StaleDeviceLists(u.process.Context(), []gomatrixserverlib.ServerName{})
if err != nil {
return err
}
res := rsapi.QueryLeftUsersResponse{}
if err = u.rsAPI.QueryLeftUsers(u.process.Context(), &rsapi.QueryLeftUsersRequest{StaleDeviceListUsers: staleUsers}, &res); err != nil {
return err
}
if len(res.LeftUsers) == 0 {
return nil
}
logrus.Debugf("Deleting %d stale device list entries", len(res.LeftUsers))
return u.db.DeleteStaleDeviceLists(u.process.Context(), res.LeftUsers)
}
func (u *DeviceListUpdater) mutex(userID string) *sync.Mutex {
u.mu.Lock()
defer u.mu.Unlock()

86
keyserver/internal/device_list_update_test.go
View file

@ -30,7 +30,12 @@ import (
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/dendrite/keyserver/api"
"github.com/matrix-org/dendrite/keyserver/storage"
roomserver "github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/process"
"github.com/matrix-org/dendrite/test"
"github.com/matrix-org/dendrite/test/testrig"
)
var (
@ -53,6 +58,10 @@ type mockDeviceListUpdaterDatabase struct {
mu sync.Mutex // protect staleUsers
}
func (d *mockDeviceListUpdaterDatabase) DeleteStaleDeviceLists(ctx context.Context, userIDs []string) error {
return nil
}
// StaleDeviceLists returns a list of user IDs ending with the domains provided who have stale device lists.
// If no domains are given, all user IDs with stale device lists are returned.
func (d *mockDeviceListUpdaterDatabase) StaleDeviceLists(ctx context.Context, domains []gomatrixserverlib.ServerName) ([]string, error) {
@ -153,7 +162,7 @@ func TestUpdateHavePrevID(t *testing.T) {
}
ap := &mockDeviceListUpdaterAPI{}
producer := &mockKeyChangeProducer{}
updater := NewDeviceListUpdater(process.NewProcessContext(), db, ap, producer, nil, 1, "localhost")
updater := NewDeviceListUpdater(process.NewProcessContext(), db, ap, producer, nil, 1, nil, "localhost")
event := gomatrixserverlib.DeviceListUpdateEvent{
DeviceDisplayName: "Foo Bar",
Deleted: false,
@ -225,7 +234,7 @@ func TestUpdateNoPrevID(t *testing.T) {
`)),
}, nil
})
updater := NewDeviceListUpdater(process.NewProcessContext(), db, ap, producer, fedClient, 2, "example.test")
updater := NewDeviceListUpdater(process.NewProcessContext(), db, ap, producer, fedClient, 2, nil, "example.test")
if err := updater.Start(); err != nil {
t.Fatalf("failed to start updater: %s", err)
}
@ -239,6 +248,7 @@ func TestUpdateNoPrevID(t *testing.T) {
UserID: remoteUserID,
}
err := updater.Update(ctx, event)
if err != nil {
t.Fatalf("Update returned an error: %s", err)
}
@ -294,7 +304,7 @@ func TestDebounce(t *testing.T) {
close(incomingFedReq)
return <-fedCh, nil
})
updater := NewDeviceListUpdater(process.NewProcessContext(), db, ap, producer, fedClient, 1, "localhost")
updater := NewDeviceListUpdater(process.NewProcessContext(), db, ap, producer, fedClient, 1, nil, "localhost")
if err := updater.Start(); err != nil {
t.Fatalf("failed to start updater: %s", err)
}
@ -349,3 +359,73 @@ func TestDebounce(t *testing.T) {
t.Errorf("user %s is marked as stale", userID)
}
}
func mustCreateKeyserverDB(t *testing.T, dbType test.DBType) (storage.Database, func()) {
t.Helper()
base, _, _ := testrig.Base(nil)
connStr, clearDB := test.PrepareDBConnectionString(t, dbType)
db, err := storage.NewDatabase(base, &config.DatabaseOptions{ConnectionString: config.DataSource(connStr)})
if err != nil {
t.Fatal(err)
}
return db, clearDB
}
type mockKeyserverRoomserverAPI struct {
leftUsers []string
}
func (m *mockKeyserverRoomserverAPI) QueryLeftUsers(ctx context.Context, req *roomserver.QueryLeftUsersRequest, res *roomserver.QueryLeftUsersResponse) error {
res.LeftUsers = m.leftUsers
return nil
}
func TestDeviceListUpdater_CleanUp(t *testing.T) {
processCtx := process.NewProcessContext()
alice := test.NewUser(t)
bob := test.NewUser(t)
// Bob is not joined to any of our rooms
rsAPI := &mockKeyserverRoomserverAPI{leftUsers: []string{bob.ID}}
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
db, clearDB := mustCreateKeyserverDB(t, dbType)
defer clearDB()
// This should not get deleted
if err := db.MarkDeviceListStale(processCtx.Context(), alice.ID, true); err != nil {
t.Error(err)
}
// this one should get deleted
if err := db.MarkDeviceListStale(processCtx.Context(), bob.ID, true); err != nil {
t.Error(err)
}
updater := NewDeviceListUpdater(processCtx, db, nil,
nil, nil,
0, rsAPI, "test")
if err := updater.CleanUp(); err != nil {
t.Error(err)
}
// check that we still have Alice in our stale list
staleUsers, err := db.StaleDeviceLists(ctx, []gomatrixserverlib.ServerName{"test"})
if err != nil {
t.Error(err)
}
// There should only be Alice
wantCount := 1
if count := len(staleUsers); count != wantCount {
t.Fatalf("expected there to be %d stale device lists, got %d", wantCount, count)
}
if staleUsers[0] != alice.ID {
t.Fatalf("unexpected stale device list user: %s, want %s", staleUsers[0], alice.ID)
}
})
}

24
keyserver/inthttp/server.go
View file

@ -21,59 +21,59 @@ import (
"github.com/matrix-org/dendrite/keyserver/api"
)
func AddRoutes(internalAPIMux *mux.Router, s api.KeyInternalAPI) {
func AddRoutes(internalAPIMux *mux.Router, s api.KeyInternalAPI, enableMetrics bool) {
internalAPIMux.Handle(
PerformClaimKeysPath,
httputil.MakeInternalRPCAPI("KeyserverPerformClaimKeys", s.PerformClaimKeys),
httputil.MakeInternalRPCAPI("KeyserverPerformClaimKeys", enableMetrics, s.PerformClaimKeys),
)
internalAPIMux.Handle(
PerformDeleteKeysPath,
httputil.MakeInternalRPCAPI("KeyserverPerformDeleteKeys", s.PerformDeleteKeys),
httputil.MakeInternalRPCAPI("KeyserverPerformDeleteKeys", enableMetrics, s.PerformDeleteKeys),
)
internalAPIMux.Handle(
PerformUploadKeysPath,
httputil.MakeInternalRPCAPI("KeyserverPerformUploadKeys", s.PerformUploadKeys),
httputil.MakeInternalRPCAPI("KeyserverPerformUploadKeys", enableMetrics, s.PerformUploadKeys),
)
internalAPIMux.Handle(
PerformUploadDeviceKeysPath,
httputil.MakeInternalRPCAPI("KeyserverPerformUploadDeviceKeys", s.PerformUploadDeviceKeys),
httputil.MakeInternalRPCAPI("KeyserverPerformUploadDeviceKeys", enableMetrics, s.PerformUploadDeviceKeys),
)
internalAPIMux.Handle(
PerformUploadDeviceSignaturesPath,
httputil.MakeInternalRPCAPI("KeyserverPerformUploadDeviceSignatures", s.PerformUploadDeviceSignatures),
httputil.MakeInternalRPCAPI("KeyserverPerformUploadDeviceSignatures", enableMetrics, s.PerformUploadDeviceSignatures),
)
internalAPIMux.Handle(
QueryKeysPath,
httputil.MakeInternalRPCAPI("KeyserverQueryKeys", s.QueryKeys),
httputil.MakeInternalRPCAPI("KeyserverQueryKeys", enableMetrics, s.QueryKeys),
)
internalAPIMux.Handle(
QueryOneTimeKeysPath,
httputil.MakeInternalRPCAPI("KeyserverQueryOneTimeKeys", s.QueryOneTimeKeys),
httputil.MakeInternalRPCAPI("KeyserverQueryOneTimeKeys", enableMetrics, s.QueryOneTimeKeys),
)
internalAPIMux.Handle(
QueryDeviceMessagesPath,
httputil.MakeInternalRPCAPI("KeyserverQueryDeviceMessages", s.QueryDeviceMessages),
httputil.MakeInternalRPCAPI("KeyserverQueryDeviceMessages", enableMetrics, s.QueryDeviceMessages),
)
internalAPIMux.Handle(
QueryKeyChangesPath,
httputil.MakeInternalRPCAPI("KeyserverQueryKeyChanges", s.QueryKeyChanges),
httputil.MakeInternalRPCAPI("KeyserverQueryKeyChanges", enableMetrics, s.QueryKeyChanges),
)
internalAPIMux.Handle(
QuerySignaturesPath,
httputil.MakeInternalRPCAPI("KeyserverQuerySignatures", s.QuerySignatures),
httputil.MakeInternalRPCAPI("KeyserverQuerySignatures", enableMetrics, s.QuerySignatures),
)
internalAPIMux.Handle(
PerformMarkAsStalePath,
httputil.MakeInternalRPCAPI("KeyserverMarkAsStale", s.PerformMarkAsStaleIfNeeded),
httputil.MakeInternalRPCAPI("KeyserverMarkAsStale", enableMetrics, s.PerformMarkAsStaleIfNeeded),
)
}

16
keyserver/keyserver.go
View file

@ -18,6 +18,8 @@ import (
"github.com/gorilla/mux"
"github.com/sirupsen/logrus"
rsapi "github.com/matrix-org/dendrite/roomserver/api"
fedsenderapi "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/keyserver/api"
"github.com/matrix-org/dendrite/keyserver/consumers"
@ -32,14 +34,15 @@ import (
// AddInternalRoutes registers HTTP handlers for the internal API. Invokes functions
// on the given input API.
func AddInternalRoutes(router *mux.Router, intAPI api.KeyInternalAPI) {
inthttp.AddRoutes(router, intAPI)
func AddInternalRoutes(router *mux.Router, intAPI api.KeyInternalAPI, enableMetrics bool) {
inthttp.AddRoutes(router, intAPI, enableMetrics)
}
// NewInternalAPI returns a concerete implementation of the internal API. Callers
// can call functions directly on the returned API or via an HTTP interface using AddInternalRoutes.
func NewInternalAPI(
base *base.BaseDendrite, cfg *config.KeyServer, fedClient fedsenderapi.KeyserverFederationAPI,
rsAPI rsapi.KeyserverRoomserverAPI,
) api.KeyInternalAPI {
js, _ := base.NATS.Prepare(base.ProcessContext, &cfg.Matrix.JetStream)
@ -47,6 +50,7 @@ func NewInternalAPI(
if err != nil {
logrus.WithError(err).Panicf("failed to connect to key server database")
}
keyChangeProducer := &producers.KeyChange{
Topic: string(cfg.Matrix.JetStream.Prefixed(jetstream.OutputKeyChangeEvent)),
JetStream: js,
@ -58,8 +62,14 @@ func NewInternalAPI(
FedClient: fedClient,
Producer: keyChangeProducer,
}
updater := internal.NewDeviceListUpdater(base.ProcessContext, db, ap, keyChangeProducer, fedClient, 8, cfg.Matrix.ServerName) // 8 workers TODO: configurable
updater := internal.NewDeviceListUpdater(base.ProcessContext, db, ap, keyChangeProducer, fedClient, 8, rsAPI, cfg.Matrix.ServerName) // 8 workers TODO: configurable
ap.Updater = updater
// Remove users which we don't share a room with anymore
if err := updater.CleanUp(); err != nil {
logrus.WithError(err).Error("failed to cleanup stale device lists")
}
go func() {
if err := updater.Start(); err != nil {
logrus.WithError(err).Panicf("failed to start device list updater")

29
keyserver/keyserver_test.go Normal file
View file

@ -0,0 +1,29 @@
package keyserver
import (
"context"
"testing"
roomserver "github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/test"
"github.com/matrix-org/dendrite/test/testrig"
)
type mockKeyserverRoomserverAPI struct {
leftUsers []string
}
func (m *mockKeyserverRoomserverAPI) QueryLeftUsers(ctx context.Context, req *roomserver.QueryLeftUsersRequest, res *roomserver.QueryLeftUsersResponse) error {
res.LeftUsers = m.leftUsers
return nil
}
// Merely tests that we can create an internal keyserver API
func Test_NewInternalAPI(t *testing.T) {
rsAPI := &mockKeyserverRoomserverAPI{}
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
base, closeBase := testrig.CreateBaseDendrite(t, dbType)
defer closeBase()
_ = NewInternalAPI(base, &base.Cfg.KeyServer, nil, rsAPI)
})
}

5
keyserver/storage/interface.go
View file

@ -85,4 +85,9 @@ type Database interface {
StoreCrossSigningKeysForUser(ctx context.Context, userID string, keyMap types.CrossSigningKeyMap) error
StoreCrossSigningSigsForTarget(ctx context.Context, originUserID string, originKeyID gomatrixserverlib.KeyID, targetUserID string, targetKeyID gomatrixserverlib.KeyID, signature gomatrixserverlib.Base64Bytes) error
DeleteStaleDeviceLists(
ctx context.Context,
userIDs []string,
) error
}

33
keyserver/storage/postgres/stale_device_lists.go
View file

@ -19,6 +19,10 @@ import (
"database/sql"
"time"
"github.com/lib/pq"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/keyserver/storage/tables"
"github.com/matrix-org/gomatrixserverlib"
@ -48,10 +52,14 @@ const selectStaleDeviceListsWithDomainsSQL = "" +
const selectStaleDeviceListsSQL = "" +
"SELECT user_id FROM keyserver_stale_device_lists WHERE is_stale = $1 ORDER BY ts_added_secs DESC"
const deleteStaleDevicesSQL = "" +
"DELETE FROM keyserver_stale_device_lists WHERE user_id = ANY($1)"
type staleDeviceListsStatements struct {
upsertStaleDeviceListStmt *sql.Stmt
selectStaleDeviceListsWithDomainsStmt *sql.Stmt
selectStaleDeviceListsStmt *sql.Stmt
deleteStaleDeviceListsStmt *sql.Stmt
}
func NewPostgresStaleDeviceListsTable(db *sql.DB) (tables.StaleDeviceLists, error) {
@ -60,16 +68,12 @@ func NewPostgresStaleDeviceListsTable(db *sql.DB) (tables.StaleDeviceLists, erro
if err != nil {
return nil, err
}
if s.upsertStaleDeviceListStmt, err = db.Prepare(upsertStaleDeviceListSQL); err != nil {
return nil, err
}
if s.selectStaleDeviceListsStmt, err = db.Prepare(selectStaleDeviceListsSQL); err != nil {
return nil, err
}
if s.selectStaleDeviceListsWithDomainsStmt, err = db.Prepare(selectStaleDeviceListsWithDomainsSQL); err != nil {
return nil, err
}
return s, nil
return s, sqlutil.StatementList{
{&s.upsertStaleDeviceListStmt, upsertStaleDeviceListSQL},
{&s.selectStaleDeviceListsStmt, selectStaleDeviceListsSQL},
{&s.selectStaleDeviceListsWithDomainsStmt, selectStaleDeviceListsWithDomainsSQL},
{&s.deleteStaleDeviceListsStmt, deleteStaleDevicesSQL},
}.Prepare(db)
}
func (s *staleDeviceListsStatements) InsertStaleDeviceList(ctx context.Context, userID string, isStale bool) error {
@ -105,6 +109,15 @@ func (s *staleDeviceListsStatements) SelectUserIDsWithStaleDeviceLists(ctx conte
return result, nil
}
// DeleteStaleDeviceLists removes users from stale device lists
func (s *staleDeviceListsStatements) DeleteStaleDeviceLists(
ctx context.Context, txn *sql.Tx, userIDs []string,
) error {
stmt := sqlutil.TxStmt(txn, s.deleteStaleDeviceListsStmt)
_, err := stmt.ExecContext(ctx, pq.Array(userIDs))
return err
}
func rowsToUserIDs(ctx context.Context, rows *sql.Rows) (result []string, err error) {
defer internal.CloseAndLogIfError(ctx, rows, "closing rowsToUserIDs failed")
for rows.Next() {

10
keyserver/storage/shared/storage.go
View file

@ -249,3 +249,13 @@ func (d *Database) StoreCrossSigningSigsForTarget(
return nil
})
}
// DeleteStaleDeviceLists deletes stale device list entries for users we don't share a room with anymore.
func (d *Database) DeleteStaleDeviceLists(
ctx context.Context,
userIDs []string,
) error {
return d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
return d.StaleDeviceListsTable.DeleteStaleDeviceLists(ctx, txn, userIDs)
})
}

44
keyserver/storage/sqlite3/stale_device_lists.go
View file

@ -17,8 +17,11 @@ package sqlite3
import (
"context"
"database/sql"
"strings"
"time"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/keyserver/storage/tables"
"github.com/matrix-org/gomatrixserverlib"
@ -48,11 +51,15 @@ const selectStaleDeviceListsWithDomainsSQL = "" +
const selectStaleDeviceListsSQL = "" +
"SELECT user_id FROM keyserver_stale_device_lists WHERE is_stale = $1 ORDER BY ts_added_secs DESC"
const deleteStaleDevicesSQL = "" +
"DELETE FROM keyserver_stale_device_lists WHERE user_id IN ($1)"
type staleDeviceListsStatements struct {
db *sql.DB
upsertStaleDeviceListStmt *sql.Stmt
selectStaleDeviceListsWithDomainsStmt *sql.Stmt
selectStaleDeviceListsStmt *sql.Stmt
// deleteStaleDeviceListsStmt *sql.Stmt // Prepared at runtime
}
func NewSqliteStaleDeviceListsTable(db *sql.DB) (tables.StaleDeviceLists, error) {
@ -63,16 +70,12 @@ func NewSqliteStaleDeviceListsTable(db *sql.DB) (tables.StaleDeviceLists, error)
if err != nil {
return nil, err
}
if s.upsertStaleDeviceListStmt, err = db.Prepare(upsertStaleDeviceListSQL); err != nil {
return nil, err
}
if s.selectStaleDeviceListsStmt, err = db.Prepare(selectStaleDeviceListsSQL); err != nil {
return nil, err
}
if s.selectStaleDeviceListsWithDomainsStmt, err = db.Prepare(selectStaleDeviceListsWithDomainsSQL); err != nil {
return nil, err
}
return s, nil
return s, sqlutil.StatementList{
{&s.upsertStaleDeviceListStmt, upsertStaleDeviceListSQL},
{&s.selectStaleDeviceListsStmt, selectStaleDeviceListsSQL},
{&s.selectStaleDeviceListsWithDomainsStmt, selectStaleDeviceListsWithDomainsSQL},
// { &s.deleteStaleDeviceListsStmt, deleteStaleDevicesSQL}, // Prepared at runtime
}.Prepare(db)
}
func (s *staleDeviceListsStatements) InsertStaleDeviceList(ctx context.Context, userID string, isStale bool) error {
@ -108,6 +111,27 @@ func (s *staleDeviceListsStatements) SelectUserIDsWithStaleDeviceLists(ctx conte
return result, nil
}
// DeleteStaleDeviceLists removes users from stale device lists
func (s *staleDeviceListsStatements) DeleteStaleDeviceLists(
ctx context.Context, txn *sql.Tx, userIDs []string,
) error {
qry := strings.Replace(deleteStaleDevicesSQL, "($1)", sqlutil.QueryVariadic(len(userIDs)), 1)
stmt, err := s.db.Prepare(qry)
if err != nil {
return err
}
defer internal.CloseAndLogIfError(ctx, stmt, "DeleteStaleDeviceLists: stmt.Close failed")
stmt = sqlutil.TxStmt(txn, stmt)
params := make([]any, len(userIDs))
for i := range userIDs {
params[i] = userIDs[i]
}
_, err = stmt.ExecContext(ctx, params...)
return err
}
func rowsToUserIDs(ctx context.Context, rows *sql.Rows) (result []string, err error) {
defer internal.CloseAndLogIfError(ctx, rows, "closing rowsToUserIDs failed")
for rows.Next() {

1
keyserver/storage/tables/interface.go
View file

@ -56,6 +56,7 @@ type KeyChanges interface {
type StaleDeviceLists interface {
InsertStaleDeviceList(ctx context.Context, userID string, isStale bool) error
SelectUserIDsWithStaleDeviceLists(ctx context.Context, domains []gomatrixserverlib.ServerName) ([]string, error)
DeleteStaleDeviceLists(ctx context.Context, txn *sql.Tx, userIDs []string) error
}
type CrossSigningKeys interface {

94
keyserver/storage/tables/stale_device_lists_test.go Normal file
View file

@ -0,0 +1,94 @@
package tables_test
import (
"context"
"testing"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/keyserver/storage/sqlite3"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/keyserver/storage/postgres"
"github.com/matrix-org/dendrite/keyserver/storage/tables"
"github.com/matrix-org/dendrite/test"
)
func mustCreateTable(t *testing.T, dbType test.DBType) (tab tables.StaleDeviceLists, close func()) {
connStr, close := test.PrepareDBConnectionString(t, dbType)
db, err := sqlutil.Open(&config.DatabaseOptions{
ConnectionString: config.DataSource(connStr),
}, nil)
if err != nil {
t.Fatalf("failed to open database: %s", err)
}
switch dbType {
case test.DBTypePostgres:
tab, err = postgres.NewPostgresStaleDeviceListsTable(db)
case test.DBTypeSQLite:
tab, err = sqlite3.NewSqliteStaleDeviceListsTable(db)
}
if err != nil {
t.Fatalf("failed to create new table: %s", err)
}
return tab, close
}
func TestStaleDeviceLists(t *testing.T) {
alice := test.NewUser(t)
bob := test.NewUser(t)
charlie := "@charlie:localhost"
ctx := context.Background()
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
tab, closeDB := mustCreateTable(t, dbType)
defer closeDB()
if err := tab.InsertStaleDeviceList(ctx, alice.ID, true); err != nil {
t.Fatalf("failed to insert stale device: %s", err)
}
if err := tab.InsertStaleDeviceList(ctx, bob.ID, true); err != nil {
t.Fatalf("failed to insert stale device: %s", err)
}
if err := tab.InsertStaleDeviceList(ctx, charlie, true); err != nil {
t.Fatalf("failed to insert stale device: %s", err)
}
// Query one server
wantStaleUsers := []string{alice.ID, bob.ID}
gotStaleUsers, err := tab.SelectUserIDsWithStaleDeviceLists(ctx, []gomatrixserverlib.ServerName{"test"})
if err != nil {
t.Fatalf("failed to query stale device lists: %s", err)
}
if !test.UnsortedStringSliceEqual(wantStaleUsers, gotStaleUsers) {
t.Fatalf("expected stale users %v, got %v", wantStaleUsers, gotStaleUsers)
}
// Query all servers
wantStaleUsers = []string{alice.ID, bob.ID, charlie}
gotStaleUsers, err = tab.SelectUserIDsWithStaleDeviceLists(ctx, []gomatrixserverlib.ServerName{})
if err != nil {
t.Fatalf("failed to query stale device lists: %s", err)
}
if !test.UnsortedStringSliceEqual(wantStaleUsers, gotStaleUsers) {
t.Fatalf("expected stale users %v, got %v", wantStaleUsers, gotStaleUsers)
}
// Delete stale devices
deleteUsers := []string{alice.ID, bob.ID}
if err = tab.DeleteStaleDeviceLists(ctx, nil, deleteUsers); err != nil {
t.Fatalf("failed to delete stale device lists: %s", err)
}
// Verify we don't get anything back after deleting
gotStaleUsers, err = tab.SelectUserIDsWithStaleDeviceLists(ctx, []gomatrixserverlib.ServerName{"test"})
if err != nil {
t.Fatalf("failed to query stale device lists: %s", err)
}
if gotCount := len(gotStaleUsers); gotCount > 0 {
t.Fatalf("expected no stale users, got %d", gotCount)
}
})
}

8
relayapi/inthttp/server.go
View file

@ -9,19 +9,19 @@ import (
// AddRoutes adds the RelayInternalAPI handlers to the http.ServeMux.
// nolint:gocyclo
func AddRoutes(intAPI api.RelayInternalAPI, internalAPIMux *mux.Router) {
func AddRoutes(intAPI api.RelayInternalAPI, internalAPIMux *mux.Router, enableMetrics bool) {
internalAPIMux.Handle(
RelayAPIPerformRelayServerSyncPath,
httputil.MakeInternalRPCAPI("RelayAPIPerformRelayServerSync", intAPI.PerformRelayServerSync),
httputil.MakeInternalRPCAPI("RelayAPIPerformRelayServerSync", enableMetrics, intAPI.PerformRelayServerSync),
)
internalAPIMux.Handle(
RelayAPIPerformStoreAsyncPath,
httputil.MakeInternalRPCAPI("RelayAPIPerformStoreAsync", intAPI.PerformStoreAsync),
httputil.MakeInternalRPCAPI("RelayAPIPerformStoreAsync", enableMetrics, intAPI.PerformStoreAsync),
)
internalAPIMux.Handle(
RelayAPIQueryAsyncTransactionsPath,
httputil.MakeInternalRPCAPI("RelayAPIQueryAsyncTransactions", intAPI.QueryAsyncTransactions),
httputil.MakeInternalRPCAPI("RelayAPIQueryAsyncTransactions", enableMetrics, intAPI.QueryAsyncTransactions),
)
}

4
relayapi/relayapi.go
View file

@ -30,8 +30,8 @@ import (
// AddInternalRoutes registers HTTP handlers for the internal API. Invokes functions
// on the given input API.
func AddInternalRoutes(router *mux.Router, intAPI api.RelayInternalAPI) {
inthttp.AddRoutes(intAPI, router)
func AddInternalRoutes(router *mux.Router, intAPI api.RelayInternalAPI, enableMetrics bool) {
inthttp.AddRoutes(intAPI, router, enableMetrics)
}
// AddPublicRoutes sets up and registers HTTP handlers on the base API muxes for the FederationAPI component.

2
relayapi/relayapi_test.go
View file

@ -58,7 +58,7 @@ func TestCreateRelayInternalRoutes(t *testing.T) {
relayAPI := relayapi.NewRelayInternalAPI(base, nil, nil, nil, nil)
assert.NotNil(t, relayAPI)
relayapi.AddInternalRoutes(base.InternalAPIMux, &relayAPI)
relayapi.AddInternalRoutes(base.InternalAPIMux, &relayAPI, false)
}
func TestCreateInvalidRelayPublicRoutesPanics(t *testing.T) {

5
roomserver/api/api.go
View file

@ -17,6 +17,7 @@ type RoomserverInternalAPI interface {
ClientRoomserverAPI
UserRoomserverAPI
FederationRoomserverAPI
KeyserverRoomserverAPI
// needed to avoid chicken and egg scenario when setting up the
// interdependencies between the roomserver and other input APIs
@ -199,3 +200,7 @@ type FederationRoomserverAPI interface {
// Query a given amount (or less) of events prior to a given set of events.
PerformBackfill(ctx context.Context, req *PerformBackfillRequest, res *PerformBackfillResponse) error
}
type KeyserverRoomserverAPI interface {
QueryLeftUsers(ctx context.Context, req *QueryLeftUsersRequest, res *QueryLeftUsersResponse) error
}

6
roomserver/api/api_trace.go
View file

@ -19,6 +19,12 @@ type RoomserverInternalAPITrace struct {
Impl RoomserverInternalAPI
}
func (t *RoomserverInternalAPITrace) QueryLeftUsers(ctx context.Context, req *QueryLeftUsersRequest, res *QueryLeftUsersResponse) error {
err := t.Impl.QueryLeftUsers(ctx, req, res)
util.GetLogger(ctx).WithError(err).Infof("QueryLeftUsers req=%+v res=%+v", js(req), js(res))
return err
}
func (t *RoomserverInternalAPITrace) SetFederationAPI(fsAPI fsAPI.RoomserverFederationAPI, keyRing *gomatrixserverlib.KeyRing) {
t.Impl.SetFederationAPI(fsAPI, keyRing)
}

12
roomserver/api/query.go
View file

@ -447,3 +447,15 @@ type QueryMembershipAtEventResponse struct {
// do not have known state will return an empty array here.
Memberships map[string][]*gomatrixserverlib.HeaderedEvent `json:"memberships"`
}
// QueryLeftUsersRequest is a request to calculate users that we (the server) don't share a
// a room with anymore. This is used to cleanup stale device list entries, where we would
// otherwise keep on trying to get device lists.
type QueryLeftUsersRequest struct {
StaleDeviceListUsers []string `json:"user_ids"`
}
// QueryLeftUsersResponse is the response to QueryLeftUsersRequest.
type QueryLeftUsersResponse struct {
LeftUsers []string `json:"user_ids"`
}

6
roomserver/internal/query/query.go
View file

@ -805,6 +805,12 @@ func (r *Queryer) QueryBulkStateContent(ctx context.Context, req *api.QueryBulkS
return nil
}
func (r *Queryer) QueryLeftUsers(ctx context.Context, req *api.QueryLeftUsersRequest, res *api.QueryLeftUsersResponse) error {
var err error
res.LeftUsers, err = r.DB.GetLeftUsers(ctx, req.StaleDeviceListUsers)
return err
}
func (r *Queryer) QuerySharedUsers(ctx context.Context, req *api.QuerySharedUsersRequest, res *api.QuerySharedUsersResponse) error {
roomIDs, err := r.DB.GetRoomsByMembership(ctx, req.UserID, "join")
if err != nil {

8
roomserver/inthttp/client.go
View file

@ -63,6 +63,7 @@ const (
RoomserverQueryAuthChainPath = "/roomserver/queryAuthChain"
RoomserverQueryRestrictedJoinAllowed = "/roomserver/queryRestrictedJoinAllowed"
RoomserverQueryMembershipAtEventPath = "/roomserver/queryMembershipAtEvent"
RoomserverQueryLeftMembersPath = "/roomserver/queryLeftMembers"
)
type httpRoomserverInternalAPI struct {
@ -553,3 +554,10 @@ func (h *httpRoomserverInternalAPI) QueryMembershipAtEvent(ctx context.Context,
h.httpClient, ctx, request, response,
)
}
func (h *httpRoomserverInternalAPI) QueryLeftUsers(ctx context.Context, request *api.QueryLeftUsersRequest, response *api.QueryLeftUsersResponse) error {
return httputil.CallInternalRPCAPI(
"RoomserverQueryLeftMembers", h.roomserverURL+RoomserverQueryLeftMembersPath,
h.httpClient, ctx, request, response,
)
}

85
roomserver/inthttp/server.go
View file

@ -9,198 +9,203 @@ import (
// AddRoutes adds the RoomserverInternalAPI handlers to the http.ServeMux.
// nolint: gocyclo
func AddRoutes(r api.RoomserverInternalAPI, internalAPIMux *mux.Router) {
func AddRoutes(r api.RoomserverInternalAPI, internalAPIMux *mux.Router, enableMetrics bool) {
internalAPIMux.Handle(
RoomserverInputRoomEventsPath,
httputil.MakeInternalRPCAPI("RoomserverInputRoomEvents", r.InputRoomEvents),
httputil.MakeInternalRPCAPI("RoomserverInputRoomEvents", enableMetrics, r.InputRoomEvents),
)
internalAPIMux.Handle(
RoomserverPerformInvitePath,
httputil.MakeInternalRPCAPI("RoomserverPerformInvite", r.PerformInvite),
httputil.MakeInternalRPCAPI("RoomserverPerformInvite", enableMetrics, r.PerformInvite),
)
internalAPIMux.Handle(
RoomserverPerformJoinPath,
httputil.MakeInternalRPCAPI("RoomserverPerformJoin", r.PerformJoin),
httputil.MakeInternalRPCAPI("RoomserverPerformJoin", enableMetrics, r.PerformJoin),
)
internalAPIMux.Handle(
RoomserverPerformLeavePath,
httputil.MakeInternalRPCAPI("RoomserverPerformLeave", r.PerformLeave),
httputil.MakeInternalRPCAPI("RoomserverPerformLeave", enableMetrics, r.PerformLeave),
)
internalAPIMux.Handle(
RoomserverPerformPeekPath,
httputil.MakeInternalRPCAPI("RoomserverPerformPeek", r.PerformPeek),
httputil.MakeInternalRPCAPI("RoomserverPerformPeek", enableMetrics, r.PerformPeek),
)
internalAPIMux.Handle(
RoomserverPerformInboundPeekPath,
httputil.MakeInternalRPCAPI("RoomserverPerformInboundPeek", r.PerformInboundPeek),
httputil.MakeInternalRPCAPI("RoomserverPerformInboundPeek", enableMetrics, r.PerformInboundPeek),
)
internalAPIMux.Handle(
RoomserverPerformUnpeekPath,
httputil.MakeInternalRPCAPI("RoomserverPerformUnpeek", r.PerformUnpeek),
httputil.MakeInternalRPCAPI("RoomserverPerformUnpeek", enableMetrics, r.PerformUnpeek),
)
internalAPIMux.Handle(
RoomserverPerformRoomUpgradePath,
httputil.MakeInternalRPCAPI("RoomserverPerformRoomUpgrade", r.PerformRoomUpgrade),
httputil.MakeInternalRPCAPI("RoomserverPerformRoomUpgrade", enableMetrics, r.PerformRoomUpgrade),
)
internalAPIMux.Handle(
RoomserverPerformPublishPath,
httputil.MakeInternalRPCAPI("RoomserverPerformPublish", r.PerformPublish),
httputil.MakeInternalRPCAPI("RoomserverPerformPublish", enableMetrics, r.PerformPublish),
)
internalAPIMux.Handle(
RoomserverPerformAdminEvacuateRoomPath,
httputil.MakeInternalRPCAPI("RoomserverPerformAdminEvacuateRoom", r.PerformAdminEvacuateRoom),
httputil.MakeInternalRPCAPI("RoomserverPerformAdminEvacuateRoom", enableMetrics, r.PerformAdminEvacuateRoom),
)
internalAPIMux.Handle(
RoomserverPerformAdminEvacuateUserPath,
httputil.MakeInternalRPCAPI("RoomserverPerformAdminEvacuateUser", r.PerformAdminEvacuateUser),
httputil.MakeInternalRPCAPI("RoomserverPerformAdminEvacuateUser", enableMetrics, r.PerformAdminEvacuateUser),
)
internalAPIMux.Handle(
RoomserverPerformAdminDownloadStatePath,
httputil.MakeInternalRPCAPI("RoomserverPerformAdminDownloadState", r.PerformAdminDownloadState),
httputil.MakeInternalRPCAPI("RoomserverPerformAdminDownloadState", enableMetrics, r.PerformAdminDownloadState),
)
internalAPIMux.Handle(
RoomserverQueryPublishedRoomsPath,
httputil.MakeInternalRPCAPI("RoomserverQueryPublishedRooms", r.QueryPublishedRooms),
httputil.MakeInternalRPCAPI("RoomserverQueryPublishedRooms", enableMetrics, r.QueryPublishedRooms),
)
internalAPIMux.Handle(
RoomserverQueryLatestEventsAndStatePath,
httputil.MakeInternalRPCAPI("RoomserverQueryLatestEventsAndState", r.QueryLatestEventsAndState),
httputil.MakeInternalRPCAPI("RoomserverQueryLatestEventsAndState", enableMetrics, r.QueryLatestEventsAndState),
)
internalAPIMux.Handle(
RoomserverQueryStateAfterEventsPath,
httputil.MakeInternalRPCAPI("RoomserverQueryStateAfterEvents", r.QueryStateAfterEvents),
httputil.MakeInternalRPCAPI("RoomserverQueryStateAfterEvents", enableMetrics, r.QueryStateAfterEvents),
)
internalAPIMux.Handle(
RoomserverQueryEventsByIDPath,
httputil.MakeInternalRPCAPI("RoomserverQueryEventsByID", r.QueryEventsByID),
httputil.MakeInternalRPCAPI("RoomserverQueryEventsByID", enableMetrics, r.QueryEventsByID),
)
internalAPIMux.Handle(
RoomserverQueryMembershipForUserPath,
httputil.MakeInternalRPCAPI("RoomserverQueryMembershipForUser", r.QueryMembershipForUser),
httputil.MakeInternalRPCAPI("RoomserverQueryMembershipForUser", enableMetrics, r.QueryMembershipForUser),
)
internalAPIMux.Handle(
RoomserverQueryMembershipsForRoomPath,
httputil.MakeInternalRPCAPI("RoomserverQueryMembershipsForRoom", r.QueryMembershipsForRoom),
httputil.MakeInternalRPCAPI("RoomserverQueryMembershipsForRoom", enableMetrics, r.QueryMembershipsForRoom),
)
internalAPIMux.Handle(
RoomserverQueryServerJoinedToRoomPath,
httputil.MakeInternalRPCAPI("RoomserverQueryServerJoinedToRoom", r.QueryServerJoinedToRoom),
httputil.MakeInternalRPCAPI("RoomserverQueryServerJoinedToRoom", enableMetrics, r.QueryServerJoinedToRoom),
)
internalAPIMux.Handle(
RoomserverQueryServerAllowedToSeeEventPath,
httputil.MakeInternalRPCAPI("RoomserverQueryServerAllowedToSeeEvent", r.QueryServerAllowedToSeeEvent),
httputil.MakeInternalRPCAPI("RoomserverQueryServerAllowedToSeeEvent", enableMetrics, r.QueryServerAllowedToSeeEvent),
)
internalAPIMux.Handle(
RoomserverQueryMissingEventsPath,
httputil.MakeInternalRPCAPI("RoomserverQueryMissingEvents", r.QueryMissingEvents),
httputil.MakeInternalRPCAPI("RoomserverQueryMissingEvents", enableMetrics, r.QueryMissingEvents),
)
internalAPIMux.Handle(
RoomserverQueryStateAndAuthChainPath,
httputil.MakeInternalRPCAPI("RoomserverQueryStateAndAuthChain", r.QueryStateAndAuthChain),
httputil.MakeInternalRPCAPI("RoomserverQueryStateAndAuthChain", enableMetrics, r.QueryStateAndAuthChain),
)
internalAPIMux.Handle(
RoomserverPerformBackfillPath,
httputil.MakeInternalRPCAPI("RoomserverPerformBackfill", r.PerformBackfill),
httputil.MakeInternalRPCAPI("RoomserverPerformBackfill", enableMetrics, r.PerformBackfill),
)
internalAPIMux.Handle(
RoomserverPerformForgetPath,
httputil.MakeInternalRPCAPI("RoomserverPerformForget", r.PerformForget),
httputil.MakeInternalRPCAPI("RoomserverPerformForget", enableMetrics, r.PerformForget),
)
internalAPIMux.Handle(
RoomserverQueryRoomVersionCapabilitiesPath,
httputil.MakeInternalRPCAPI("RoomserverQueryRoomVersionCapabilities", r.QueryRoomVersionCapabilities),
httputil.MakeInternalRPCAPI("RoomserverQueryRoomVersionCapabilities", enableMetrics, r.QueryRoomVersionCapabilities),
)
internalAPIMux.Handle(
RoomserverQueryRoomVersionForRoomPath,
httputil.MakeInternalRPCAPI("RoomserverQueryRoomVersionForRoom", r.QueryRoomVersionForRoom),
httputil.MakeInternalRPCAPI("RoomserverQueryRoomVersionForRoom", enableMetrics, r.QueryRoomVersionForRoom),
)
internalAPIMux.Handle(
RoomserverSetRoomAliasPath,
httputil.MakeInternalRPCAPI("RoomserverSetRoomAlias", r.SetRoomAlias),
httputil.MakeInternalRPCAPI("RoomserverSetRoomAlias", enableMetrics, r.SetRoomAlias),
)
internalAPIMux.Handle(
RoomserverGetRoomIDForAliasPath,
httputil.MakeInternalRPCAPI("RoomserverGetRoomIDForAlias", r.GetRoomIDForAlias),
httputil.MakeInternalRPCAPI("RoomserverGetRoomIDForAlias", enableMetrics, r.GetRoomIDForAlias),
)
internalAPIMux.Handle(
RoomserverGetAliasesForRoomIDPath,
httputil.MakeInternalRPCAPI("RoomserverGetAliasesForRoomID", r.GetAliasesForRoomID),
httputil.MakeInternalRPCAPI("RoomserverGetAliasesForRoomID", enableMetrics, r.GetAliasesForRoomID),
)
internalAPIMux.Handle(
RoomserverRemoveRoomAliasPath,
httputil.MakeInternalRPCAPI("RoomserverRemoveRoomAlias", r.RemoveRoomAlias),
httputil.MakeInternalRPCAPI("RoomserverRemoveRoomAlias", enableMetrics, r.RemoveRoomAlias),
)
internalAPIMux.Handle(
RoomserverQueryCurrentStatePath,
httputil.MakeInternalRPCAPI("RoomserverQueryCurrentState", r.QueryCurrentState),
httputil.MakeInternalRPCAPI("RoomserverQueryCurrentState", enableMetrics, r.QueryCurrentState),
)
internalAPIMux.Handle(
RoomserverQueryRoomsForUserPath,
httputil.MakeInternalRPCAPI("RoomserverQueryRoomsForUser", r.QueryRoomsForUser),
httputil.MakeInternalRPCAPI("RoomserverQueryRoomsForUser", enableMetrics, r.QueryRoomsForUser),
)
internalAPIMux.Handle(
RoomserverQueryBulkStateContentPath,
httputil.MakeInternalRPCAPI("RoomserverQueryBulkStateContent", r.QueryBulkStateContent),
httputil.MakeInternalRPCAPI("RoomserverQueryBulkStateContent", enableMetrics, r.QueryBulkStateContent),
)
internalAPIMux.Handle(
RoomserverQuerySharedUsersPath,
httputil.MakeInternalRPCAPI("RoomserverQuerySharedUsers", r.QuerySharedUsers),
httputil.MakeInternalRPCAPI("RoomserverQuerySharedUsers", enableMetrics, r.QuerySharedUsers),
)
internalAPIMux.Handle(
RoomserverQueryKnownUsersPath,
httputil.MakeInternalRPCAPI("RoomserverQueryKnownUsers", r.QueryKnownUsers),
httputil.MakeInternalRPCAPI("RoomserverQueryKnownUsers", enableMetrics, r.QueryKnownUsers),
)
internalAPIMux.Handle(
RoomserverQueryServerBannedFromRoomPath,
httputil.MakeInternalRPCAPI("RoomserverQueryServerBannedFromRoom", r.QueryServerBannedFromRoom),
httputil.MakeInternalRPCAPI("RoomserverQueryServerBannedFromRoom", enableMetrics, r.QueryServerBannedFromRoom),
)
internalAPIMux.Handle(
RoomserverQueryAuthChainPath,
httputil.MakeInternalRPCAPI("RoomserverQueryAuthChain", r.QueryAuthChain),
httputil.MakeInternalRPCAPI("RoomserverQueryAuthChain", enableMetrics, r.QueryAuthChain),
)
internalAPIMux.Handle(
RoomserverQueryRestrictedJoinAllowed,
httputil.MakeInternalRPCAPI("RoomserverQueryRestrictedJoinAllowed", r.QueryRestrictedJoinAllowed),
httputil.MakeInternalRPCAPI("RoomserverQueryRestrictedJoinAllowed", enableMetrics, r.QueryRestrictedJoinAllowed),
)
internalAPIMux.Handle(
RoomserverQueryMembershipAtEventPath,
httputil.MakeInternalRPCAPI("RoomserverQueryMembershipAtEventPath", r.QueryMembershipAtEvent),
httputil.MakeInternalRPCAPI("RoomserverQueryMembershipAtEventPath", enableMetrics, r.QueryMembershipAtEvent),
)
internalAPIMux.Handle(
RoomserverQueryLeftMembersPath,
httputil.MakeInternalRPCAPI("RoomserverQueryLeftMembersPath", enableMetrics, r.QueryLeftUsers),
)
}

7
roomserver/roomserver.go
View file

@ -16,18 +16,19 @@ package roomserver
import (
"github.com/gorilla/mux"
"github.com/sirupsen/logrus"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/internal"
"github.com/matrix-org/dendrite/roomserver/inthttp"
"github.com/matrix-org/dendrite/roomserver/storage"
"github.com/matrix-org/dendrite/setup/base"
"github.com/sirupsen/logrus"
)
// AddInternalRoutes registers HTTP handlers for the internal API. Invokes functions
// on the given input API.
func AddInternalRoutes(router *mux.Router, intAPI api.RoomserverInternalAPI) {
inthttp.AddRoutes(intAPI, router)
func AddInternalRoutes(router *mux.Router, intAPI api.RoomserverInternalAPI, enableMetrics bool) {
inthttp.AddRoutes(intAPI, router, enableMetrics)
}
// NewInternalAPI returns a concerete implementation of the internal API. Callers

77
roomserver/roomserver_test.go
View file

@ -2,20 +2,27 @@ package roomserver_test
import (
"context"
"net/http"
"testing"
"time"
"github.com/gorilla/mux"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/roomserver"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/roomserver/inthttp"
"github.com/matrix-org/dendrite/roomserver/storage"
"github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/test"
"github.com/matrix-org/dendrite/test/testrig"
"github.com/matrix-org/gomatrixserverlib"
)
func mustCreateDatabase(t *testing.T, dbType test.DBType) (*base.BaseDendrite, storage.Database, func()) {
t.Helper()
base, close := testrig.CreateBaseDendrite(t, dbType)
db, err := storage.Open(base, &base.Cfg.KeyServer.Database, base.Caches)
db, err := storage.Open(base, &base.Cfg.RoomServer.Database, base.Caches)
if err != nil {
t.Fatalf("failed to create Database: %v", err)
}
@ -67,3 +74,69 @@ func Test_SharedUsers(t *testing.T) {
}
})
}
func Test_QueryLeftUsers(t *testing.T) {
alice := test.NewUser(t)
bob := test.NewUser(t)
room := test.NewRoom(t, alice, test.RoomPreset(test.PresetTrustedPrivateChat))
// Invite and join Bob
room.CreateAndInsert(t, alice, gomatrixserverlib.MRoomMember, map[string]interface{}{
"membership": "invite",
}, test.WithStateKey(bob.ID))
room.CreateAndInsert(t, bob, gomatrixserverlib.MRoomMember, map[string]interface{}{
"membership": "join",
}, test.WithStateKey(bob.ID))
ctx := context.Background()
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
base, _, close := mustCreateDatabase(t, dbType)
defer close()
rsAPI := roomserver.NewInternalAPI(base)
// SetFederationAPI starts the room event input consumer
rsAPI.SetFederationAPI(nil, nil)
// Create the room
if err := api.SendEvents(ctx, rsAPI, api.KindNew, room.Events(), "test", "test", "test", nil, false); err != nil {
t.Fatalf("failed to send events: %v", err)
}
// Query the left users, there should only be "@idontexist:test",
// as Alice and Bob are still joined.
res := &api.QueryLeftUsersResponse{}
leftUserID := "@idontexist:test"
getLeftUsersList := []string{alice.ID, bob.ID, leftUserID}
testCase := func(rsAPI api.RoomserverInternalAPI) {
if err := rsAPI.QueryLeftUsers(ctx, &api.QueryLeftUsersRequest{StaleDeviceListUsers: getLeftUsersList}, res); err != nil {
t.Fatalf("unable to query left users: %v", err)
}
wantCount := 1
if count := len(res.LeftUsers); count > wantCount {
t.Fatalf("unexpected left users count: want %d, got %d", wantCount, count)
}
if res.LeftUsers[0] != leftUserID {
t.Fatalf("unexpected left users : want %s, got %s", leftUserID, res.LeftUsers[0])
}
}
t.Run("HTTP API", func(t *testing.T) {
router := mux.NewRouter().PathPrefix(httputil.InternalPathPrefix).Subrouter()
roomserver.AddInternalRoutes(router, rsAPI, false)
apiURL, cancel := test.ListenAndServe(t, router, false)
defer cancel()
httpAPI, err := inthttp.NewRoomserverClient(apiURL, &http.Client{Timeout: time.Second * 5}, nil)
if err != nil {
t.Fatalf("failed to create HTTP client")
}
testCase(httpAPI)
})
t.Run("Monolith", func(t *testing.T) {
testCase(rsAPI)
// also test tracing
traceAPI := &api.RoomserverInternalAPITrace{Impl: rsAPI}
testCase(traceAPI)
})
})
}

1
roomserver/storage/interface.go
View file

@ -172,5 +172,6 @@ type Database interface {
ForgetRoom(ctx context.Context, userID, roomID string, forget bool) error
GetHistoryVisibilityState(ctx context.Context, roomInfo *types.RoomInfo, eventID string, domain string) ([]*gomatrixserverlib.Event, error)
GetLeftUsers(ctx context.Context, userIDs []string) ([]string, error)
UpgradeRoom(ctx context.Context, oldRoomID, newRoomID, eventSender string) error
}

34
roomserver/storage/postgres/membership_table.go
View file

@ -21,12 +21,13 @@ import (
"fmt"
"github.com/lib/pq"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver/storage/postgres/deltas"
"github.com/matrix-org/dendrite/roomserver/storage/tables"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/gomatrixserverlib"
)
const membershipSchema = `
@ -157,6 +158,12 @@ const selectServerInRoomSQL = "" +
" JOIN roomserver_event_state_keys ON roomserver_membership.target_nid = roomserver_event_state_keys.event_state_key_nid" +
" WHERE membership_nid = $1 AND room_nid = $2 AND event_state_key LIKE '%:' || $3 LIMIT 1"
const selectJoinedUsersSQL = `
SELECT DISTINCT target_nid
FROM roomserver_membership m
WHERE membership_nid > $1 AND target_nid = ANY($2)
`
type membershipStatements struct {
insertMembershipStmt *sql.Stmt
selectMembershipForUpdateStmt *sql.Stmt
@ -174,6 +181,7 @@ type membershipStatements struct {
selectLocalServerInRoomStmt *sql.Stmt
selectServerInRoomStmt *sql.Stmt
deleteMembershipStmt *sql.Stmt
selectJoinedUsersStmt *sql.Stmt
}
func CreateMembershipTable(db *sql.DB) error {
@ -209,9 +217,33 @@ func PrepareMembershipTable(db *sql.DB) (tables.Membership, error) {
{&s.selectLocalServerInRoomStmt, selectLocalServerInRoomSQL},
{&s.selectServerInRoomStmt, selectServerInRoomSQL},
{&s.deleteMembershipStmt, deleteMembershipSQL},
{&s.selectJoinedUsersStmt, selectJoinedUsersSQL},
}.Prepare(db)
}
func (s *membershipStatements) SelectJoinedUsers(
ctx context.Context, txn *sql.Tx,
targetUserNIDs []types.EventStateKeyNID,
) ([]types.EventStateKeyNID, error) {
result := make([]types.EventStateKeyNID, 0, len(targetUserNIDs))
stmt := sqlutil.TxStmt(txn, s.selectJoinedUsersStmt)
rows, err := stmt.QueryContext(ctx, tables.MembershipStateLeaveOrBan, pq.Array(targetUserNIDs))
if err != nil {
return nil, err
}
defer internal.CloseAndLogIfError(ctx, rows, "SelectJoinedUsers: rows.close() failed")
var targetNID types.EventStateKeyNID
for rows.Next() {
if err = rows.Scan(&targetNID); err != nil {
return nil, err
}
result = append(result, targetNID)
}
return result, rows.Err()
}
func (s *membershipStatements) InsertMembership(
ctx context.Context, txn *sql.Tx,
roomNID types.RoomNID, targetUserNID types.EventStateKeyNID,

37
roomserver/storage/shared/storage.go
View file

@ -1365,6 +1365,43 @@ func (d *Database) JoinedUsersSetInRooms(ctx context.Context, roomIDs, userIDs [
return result, nil
}
// GetLeftUsers calculates users we (the server) don't share a room with anymore.
func (d *Database) GetLeftUsers(ctx context.Context, userIDs []string) ([]string, error) {
// Get the userNID for all users with a stale device list
stateKeyNIDMap, err := d.EventStateKeyNIDs(ctx, userIDs)
if err != nil {
return nil, err
}
userNIDs := make([]types.EventStateKeyNID, 0, len(stateKeyNIDMap))
userNIDtoUserID := make(map[types.EventStateKeyNID]string, len(stateKeyNIDMap))
// Create a map from userNID -> userID
for userID, nid := range stateKeyNIDMap {
userNIDs = append(userNIDs, nid)
userNIDtoUserID[nid] = userID
}
// Get all users whose membership is still join, knock or invite.
stillJoinedUsersNIDs, err := d.MembershipTable.SelectJoinedUsers(ctx, nil, userNIDs)
if err != nil {
return nil, err
}
// Remove joined users from the "user with stale devices" list, which contains left AND joined users
for _, joinedUser := range stillJoinedUsersNIDs {
delete(userNIDtoUserID, joinedUser)
}
// The users still in our userNIDtoUserID map are the users we don't share a room with anymore,
// and the return value we are looking for.
leftUsers := make([]string, 0, len(userNIDtoUserID))
for _, userID := range userNIDtoUserID {
leftUsers = append(leftUsers, userID)
}
return leftUsers, nil
}
// GetLocalServerInRoom returns true if we think we're in a given room or false otherwise.
func (d *Database) GetLocalServerInRoom(ctx context.Context, roomNID types.RoomNID) (bool, error) {
return d.MembershipTable.SelectLocalServerInRoom(ctx, nil, roomNID)

96
roomserver/storage/shared/storage_test.go Normal file
View file

@ -0,0 +1,96 @@
package shared_test
import (
"context"
"testing"
"github.com/stretchr/testify/assert"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver/storage/postgres"
"github.com/matrix-org/dendrite/roomserver/storage/shared"
"github.com/matrix-org/dendrite/roomserver/storage/sqlite3"
"github.com/matrix-org/dendrite/roomserver/storage/tables"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/test"
"github.com/matrix-org/dendrite/test/testrig"
)
func mustCreateRoomserverDatabase(t *testing.T, dbType test.DBType) (*shared.Database, func()) {
t.Helper()
connStr, clearDB := test.PrepareDBConnectionString(t, dbType)
base, _, _ := testrig.Base(nil)
dbOpts := &config.DatabaseOptions{ConnectionString: config.DataSource(connStr)}
db, err := sqlutil.Open(dbOpts, sqlutil.NewExclusiveWriter())
assert.NoError(t, err)
var membershipTable tables.Membership
var stateKeyTable tables.EventStateKeys
switch dbType {
case test.DBTypePostgres:
err = postgres.CreateEventStateKeysTable(db)
assert.NoError(t, err)
err = postgres.CreateMembershipTable(db)
assert.NoError(t, err)
membershipTable, err = postgres.PrepareMembershipTable(db)
assert.NoError(t, err)
stateKeyTable, err = postgres.PrepareEventStateKeysTable(db)
case test.DBTypeSQLite:
err = sqlite3.CreateEventStateKeysTable(db)
assert.NoError(t, err)
err = sqlite3.CreateMembershipTable(db)
assert.NoError(t, err)
membershipTable, err = sqlite3.PrepareMembershipTable(db)
assert.NoError(t, err)
stateKeyTable, err = sqlite3.PrepareEventStateKeysTable(db)
}
assert.NoError(t, err)
return &shared.Database{
DB: db,
EventStateKeysTable: stateKeyTable,
MembershipTable: membershipTable,
Writer: sqlutil.NewExclusiveWriter(),
}, func() {
err := base.Close()
assert.NoError(t, err)
clearDB()
err = db.Close()
assert.NoError(t, err)
}
}
func Test_GetLeftUsers(t *testing.T) {
alice := test.NewUser(t)
bob := test.NewUser(t)
charlie := test.NewUser(t)
ctx := context.Background()
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
db, close := mustCreateRoomserverDatabase(t, dbType)
defer close()
// Create dummy entries
for _, user := range []*test.User{alice, bob, charlie} {
nid, err := db.EventStateKeysTable.InsertEventStateKeyNID(ctx, nil, user.ID)
assert.NoError(t, err)
err = db.MembershipTable.InsertMembership(ctx, nil, 1, nid, true)
assert.NoError(t, err)
// We must update the membership with a non-zero event NID or it will get filtered out in later queries
membershipNID := tables.MembershipStateLeaveOrBan
if user == alice {
membershipNID = tables.MembershipStateJoin
}
_, err = db.MembershipTable.UpdateMembership(ctx, nil, 1, nid, nid, membershipNID, 1, false)
assert.NoError(t, err)
}
// Now try to get the left users, this should be Bob and Charlie, since they have a "leave" membership
expectedUserIDs := []string{bob.ID, charlie.ID}
leftUsers, err := db.GetLeftUsers(context.Background(), []string{alice.ID, bob.ID, charlie.ID})
assert.NoError(t, err)
assert.ElementsMatch(t, expectedUserIDs, leftUsers)
})
}

47
roomserver/storage/sqlite3/membership_table.go
View file

@ -21,12 +21,13 @@ import (
"fmt"
"strings"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver/storage/sqlite3/deltas"
"github.com/matrix-org/dendrite/roomserver/storage/tables"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/gomatrixserverlib"
)
const membershipSchema = `
@ -133,6 +134,12 @@ const selectServerInRoomSQL = "" +
const deleteMembershipSQL = "" +
"DELETE FROM roomserver_membership WHERE room_nid = $1 AND target_nid = $2"
const selectJoinedUsersSQL = `
SELECT DISTINCT target_nid
FROM roomserver_membership m
WHERE membership_nid > $1 AND target_nid IN ($2)
`
type membershipStatements struct {
db *sql.DB
insertMembershipStmt *sql.Stmt
@ -149,6 +156,7 @@ type membershipStatements struct {
selectLocalServerInRoomStmt *sql.Stmt
selectServerInRoomStmt *sql.Stmt
deleteMembershipStmt *sql.Stmt
// selectJoinedUsersStmt *sql.Stmt // Prepared at runtime
}
func CreateMembershipTable(db *sql.DB) error {
@ -412,3 +420,40 @@ func (s *membershipStatements) DeleteMembership(
)
return err
}
func (s *membershipStatements) SelectJoinedUsers(
ctx context.Context, txn *sql.Tx,
targetUserNIDs []types.EventStateKeyNID,
) ([]types.EventStateKeyNID, error) {
result := make([]types.EventStateKeyNID, 0, len(targetUserNIDs))
qry := strings.Replace(selectJoinedUsersSQL, "($2)", sqlutil.QueryVariadicOffset(len(targetUserNIDs), 1), 1)
stmt, err := s.db.Prepare(qry)
if err != nil {
return nil, err
}
defer internal.CloseAndLogIfError(ctx, stmt, "SelectJoinedUsers: stmt.Close failed")
params := make([]any, len(targetUserNIDs)+1)
params[0] = tables.MembershipStateLeaveOrBan
for i := range targetUserNIDs {
params[i+1] = targetUserNIDs[i]
}
stmt = sqlutil.TxStmt(txn, stmt)
rows, err := stmt.QueryContext(ctx, params...)
if err != nil {
return nil, err
}
defer internal.CloseAndLogIfError(ctx, rows, "SelectJoinedUsers: rows.close() failed")
var targetNID types.EventStateKeyNID
for rows.Next() {
if err = rows.Scan(&targetNID); err != nil {
return nil, err
}
result = append(result, targetNID)
}
return result, rows.Err()
}

1
roomserver/storage/tables/interface.go
View file

@ -144,6 +144,7 @@ type Membership interface {
SelectLocalServerInRoom(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID) (bool, error)
SelectServerInRoom(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID, serverName gomatrixserverlib.ServerName) (bool, error)
DeleteMembership(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID, targetUserNID types.EventStateKeyNID) error
SelectJoinedUsers(ctx context.Context, txn *sql.Tx, targetUserNIDs []types.EventStateKeyNID) ([]types.EventStateKeyNID, error)
}
type Published interface {

6
roomserver/storage/tables/membership_table_test.go
View file

@ -129,5 +129,11 @@ func TestMembershipTable(t *testing.T) {
knownUsers, err := tab.SelectKnownUsers(ctx, nil, userNIDs[0], "localhost", 2)
assert.NoError(t, err)
assert.Equal(t, 1, len(knownUsers))
// get users we share a room with, given their userNID
joinedUsers, err := tab.SelectJoinedUsers(ctx, nil, userNIDs)
assert.NoError(t, err)
// Only userNIDs[0] is actually joined, so we only expect this userNID
assert.Equal(t, userNIDs[:1], joinedUsers)
})
}

12
syncapi/consumers/presence.go
View file

@ -78,7 +78,7 @@ func (s *PresenceConsumer) Start() error {
// Normal NATS subscription, used by Request/Reply
_, err := s.nats.Subscribe(s.requestTopic, func(msg *nats.Msg) {
userID := msg.Header.Get(jetstream.UserID)
presence, err := s.db.GetPresence(context.Background(), userID)
presences, err := s.db.GetPresences(context.Background(), []string{userID})
m := &nats.Msg{
Header: nats.Header{},
}
@ -89,10 +89,12 @@ func (s *PresenceConsumer) Start() error {
}
return
}
if presence == nil {
presence = &types.PresenceInternal{
UserID: userID,
}
presence := &types.PresenceInternal{
UserID: userID,
}
if len(presences) > 0 {
presence = presences[0]
}
deviceRes := api.QueryDevicesResponse{}

4
syncapi/storage/interface.go
View file

@ -106,7 +106,7 @@ type DatabaseTransaction interface {
SelectMembershipForUser(ctx context.Context, roomID, userID string, pos int64) (membership string, topologicalPos int, err error)
// getUserUnreadNotificationCountsForRooms returns the unread notifications for the given rooms
GetUserUnreadNotificationCountsForRooms(ctx context.Context, userID string, roomIDs map[string]string) (map[string]*eventutil.NotificationData, error)
GetPresence(ctx context.Context, userID string) (*types.PresenceInternal, error)
GetPresences(ctx context.Context, userID []string) ([]*types.PresenceInternal, error)
PresenceAfter(ctx context.Context, after types.StreamPosition, filter gomatrixserverlib.EventFilter) (map[string]*types.PresenceInternal, error)
RelationsFor(ctx context.Context, roomID, eventID, relType, eventType string, from, to types.StreamPosition, backwards bool, limit int) (events []types.StreamEvent, prevBatch, nextBatch string, err error)
}
@ -186,7 +186,7 @@ type Database interface {
}
type Presence interface {
GetPresence(ctx context.Context, userID string) (*types.PresenceInternal, error)
GetPresences(ctx context.Context, userIDs []string) ([]*types.PresenceInternal, error)
UpdatePresence(ctx context.Context, userID string, presence types.Presence, statusMsg *string, lastActiveTS gomatrixserverlib.Timestamp, fromSync bool) (types.StreamPosition, error)
}

38
syncapi/storage/postgres/presence_table.go
View file

@ -19,10 +19,12 @@ import (
"database/sql"
"time"
"github.com/lib/pq"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/syncapi/types"
"github.com/matrix-org/gomatrixserverlib"
)
const presenceSchema = `
@ -63,9 +65,9 @@ const upsertPresenceFromSyncSQL = "" +
" RETURNING id"
const selectPresenceForUserSQL = "" +
"SELECT presence, status_msg, last_active_ts" +
"SELECT user_id, presence, status_msg, last_active_ts" +
" FROM syncapi_presence" +
" WHERE user_id = $1 LIMIT 1"
" WHERE user_id = ANY($1)"
const selectMaxPresenceSQL = "" +
"SELECT COALESCE(MAX(id), 0) FROM syncapi_presence"
@ -119,20 +121,28 @@ func (p *presenceStatements) UpsertPresence(
return
}
// GetPresenceForUser returns the current presence of a user.
func (p *presenceStatements) GetPresenceForUser(
// GetPresenceForUsers returns the current presence for a list of users.
// If the user doesn't have a presence status yet, it is omitted from the response.
func (p *presenceStatements) GetPresenceForUsers(
ctx context.Context, txn *sql.Tx,
userID string,
) (*types.PresenceInternal, error) {
result := &types.PresenceInternal{
UserID: userID,
}
userIDs []string,
) ([]*types.PresenceInternal, error) {
result := make([]*types.PresenceInternal, 0, len(userIDs))
stmt := sqlutil.TxStmt(txn, p.selectPresenceForUsersStmt)
err := stmt.QueryRowContext(ctx, userID).Scan(&result.Presence, &result.ClientFields.StatusMsg, &result.LastActiveTS)
if err == sql.ErrNoRows {
return nil, nil
rows, err := stmt.QueryContext(ctx, pq.Array(userIDs))
if err != nil {
return nil, err
}
defer internal.CloseAndLogIfError(ctx, rows, "GetPresenceForUsers: rows.close() failed")
for rows.Next() {
presence := &types.PresenceInternal{}
if err = rows.Scan(&presence.UserID, &presence.Presence, &presence.ClientFields.StatusMsg, &presence.LastActiveTS); err != nil {
return nil, err
}
presence.ClientFields.Presence = presence.Presence.String()
result = append(result, presence)
}
result.ClientFields.Presence = result.Presence.String()
return result, err
}

46
syncapi/storage/shared/storage_consumer.go
View file

@ -57,31 +57,23 @@ type Database struct {
}
func (d *Database) NewDatabaseSnapshot(ctx context.Context) (*DatabaseTransaction, error) {
return d.NewDatabaseTransaction(ctx)
/*
TODO: Repeatable read is probably the right thing to do here,
but it seems to cause some problems with the invite tests, so
need to investigate that further.
txn, err := d.DB.BeginTx(ctx, &sql.TxOptions{
// Set the isolation level so that we see a snapshot of the database.
// In PostgreSQL repeatable read transactions will see a snapshot taken
// at the first query, and since the transaction is read-only it can't
// run into any serialisation errors.
// https://www.postgresql.org/docs/9.5/static/transaction-iso.html#XACT-REPEATABLE-READ
Isolation: sql.LevelRepeatableRead,
ReadOnly: true,
})
if err != nil {
return nil, err
}
return &DatabaseTransaction{
Database: d,
ctx: ctx,
txn: txn,
}, nil
*/
txn, err := d.DB.BeginTx(ctx, &sql.TxOptions{
// Set the isolation level so that we see a snapshot of the database.
// In PostgreSQL repeatable read transactions will see a snapshot taken
// at the first query, and since the transaction is read-only it can't
// run into any serialisation errors.
// https://www.postgresql.org/docs/9.5/static/transaction-iso.html#XACT-REPEATABLE-READ
Isolation: sql.LevelRepeatableRead,
ReadOnly: true,
})
if err != nil {
return nil, err
}
return &DatabaseTransaction{
Database: d,
ctx: ctx,
txn: txn,
}, nil
}
func (d *Database) NewDatabaseTransaction(ctx context.Context) (*DatabaseTransaction, error) {
@ -572,8 +564,8 @@ func (d *Database) UpdatePresence(ctx context.Context, userID string, presence t
return pos, err
}
func (d *Database) GetPresence(ctx context.Context, userID string) (*types.PresenceInternal, error) {
return d.Presence.GetPresenceForUser(ctx, nil, userID)
func (d *Database) GetPresences(ctx context.Context, userIDs []string) ([]*types.PresenceInternal, error) {
return d.Presence.GetPresenceForUsers(ctx, nil, userIDs)
}
func (d *Database) SelectMembershipForUser(ctx context.Context, roomID, userID string, pos int64) (membership string, topologicalPos int, err error) {

4
syncapi/storage/shared/storage_sync.go
View file

@ -596,8 +596,8 @@ func (d *DatabaseTransaction) GetUserUnreadNotificationCountsForRooms(ctx contex
return d.NotificationData.SelectUserUnreadCountsForRooms(ctx, d.txn, userID, roomIDs)
}
func (d *DatabaseTransaction) GetPresence(ctx context.Context, userID string) (*types.PresenceInternal, error) {
return d.Presence.GetPresenceForUser(ctx, d.txn, userID)
func (d *DatabaseTransaction) GetPresences(ctx context.Context, userIDs []string) ([]*types.PresenceInternal, error) {
return d.Presence.GetPresenceForUsers(ctx, d.txn, userIDs)
}
func (d *DatabaseTransaction) PresenceAfter(ctx context.Context, after types.StreamPosition, filter gomatrixserverlib.EventFilter) (map[string]*types.PresenceInternal, error) {

48
syncapi/storage/sqlite3/presence_table.go
View file

@ -17,12 +17,14 @@ package sqlite3
import (
"context"
"database/sql"
"strings"
"time"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/syncapi/types"
"github.com/matrix-org/gomatrixserverlib"
)
const presenceSchema = `
@ -62,9 +64,9 @@ const upsertPresenceFromSyncSQL = "" +
" RETURNING id"
const selectPresenceForUserSQL = "" +
"SELECT presence, status_msg, last_active_ts" +
"SELECT user_id, presence, status_msg, last_active_ts" +
" FROM syncapi_presence" +
" WHERE user_id = $1 LIMIT 1"
" WHERE user_id IN ($1)"
const selectMaxPresenceSQL = "" +
"SELECT COALESCE(MAX(id), 0) FROM syncapi_presence"
@ -134,20 +136,38 @@ func (p *presenceStatements) UpsertPresence(
return
}
// GetPresenceForUser returns the current presence of a user.
func (p *presenceStatements) GetPresenceForUser(
// GetPresenceForUsers returns the current presence for a list of users.
// If the user doesn't have a presence status yet, it is omitted from the response.
func (p *presenceStatements) GetPresenceForUsers(
ctx context.Context, txn *sql.Tx,
userID string,
) (*types.PresenceInternal, error) {
result := &types.PresenceInternal{
UserID: userID,
userIDs []string,
) ([]*types.PresenceInternal, error) {
qry := strings.Replace(selectPresenceForUserSQL, "($1)", sqlutil.QueryVariadic(len(userIDs)), 1)
prepStmt, err := p.db.Prepare(qry)
if err != nil {
return nil, err
}
stmt := sqlutil.TxStmt(txn, p.selectPresenceForUsersStmt)
err := stmt.QueryRowContext(ctx, userID).Scan(&result.Presence, &result.ClientFields.StatusMsg, &result.LastActiveTS)
if err == sql.ErrNoRows {
return nil, nil
defer internal.CloseAndLogIfError(ctx, prepStmt, "GetPresenceForUsers: stmt.close() failed")
params := make([]interface{}, len(userIDs))
for i := range userIDs {
params[i] = userIDs[i]
}
rows, err := sqlutil.TxStmt(txn, prepStmt).QueryContext(ctx, params...)
if err != nil {
return nil, err
}
defer internal.CloseAndLogIfError(ctx, rows, "GetPresenceForUsers: rows.close() failed")
result := make([]*types.PresenceInternal, 0, len(userIDs))
for rows.Next() {
presence := &types.PresenceInternal{}
if err = rows.Scan(&presence.UserID, &presence.Presence, &presence.ClientFields.StatusMsg, &presence.LastActiveTS); err != nil {
return nil, err
}
presence.ClientFields.Presence = presence.Presence.String()
result = append(result, presence)
}
result.ClientFields.Presence = result.Presence.String()
return result, err
}

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

@ -207,7 +207,7 @@ type Ignores interface {
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)
GetPresenceForUsers(ctx context.Context, txn *sql.Tx, userIDs []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, filter gomatrixserverlib.EventFilter) (presences map[string]*types.PresenceInternal, err error)
}

136
syncapi/storage/tables/presence_table_test.go Normal file
View file

@ -0,0 +1,136 @@
package tables_test
import (
"context"
"database/sql"
"reflect"
"testing"
"time"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/syncapi/storage/postgres"
"github.com/matrix-org/dendrite/syncapi/storage/sqlite3"
"github.com/matrix-org/dendrite/syncapi/storage/tables"
"github.com/matrix-org/dendrite/syncapi/types"
"github.com/matrix-org/dendrite/test"
)
func mustPresenceTable(t *testing.T, dbType test.DBType) (tables.Presence, func()) {
t.Helper()
connStr, close := test.PrepareDBConnectionString(t, dbType)
db, err := sqlutil.Open(&config.DatabaseOptions{
ConnectionString: config.DataSource(connStr),
}, sqlutil.NewExclusiveWriter())
if err != nil {
t.Fatalf("failed to open db: %s", err)
}
var tab tables.Presence
switch dbType {
case test.DBTypePostgres:
tab, err = postgres.NewPostgresPresenceTable(db)
case test.DBTypeSQLite:
var stream sqlite3.StreamIDStatements
if err = stream.Prepare(db); err != nil {
t.Fatalf("failed to prepare stream stmts: %s", err)
}
tab, err = sqlite3.NewSqlitePresenceTable(db, &stream)
}
if err != nil {
t.Fatalf("failed to make new table: %s", err)
}
return tab, close
}
func TestPresence(t *testing.T) {
alice := test.NewUser(t)
bob := test.NewUser(t)
ctx := context.Background()
statusMsg := "Hello World!"
timestamp := gomatrixserverlib.AsTimestamp(time.Now())
var txn *sql.Tx
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
tab, closeDB := mustPresenceTable(t, dbType)
defer closeDB()
// Insert some presences
pos, err := tab.UpsertPresence(ctx, txn, alice.ID, &statusMsg, types.PresenceOnline, timestamp, false)
if err != nil {
t.Error(err)
}
wantPos := types.StreamPosition(1)
if pos != wantPos {
t.Errorf("expected pos to be %d, got %d", wantPos, pos)
}
pos, err = tab.UpsertPresence(ctx, txn, bob.ID, &statusMsg, types.PresenceOnline, timestamp, false)
if err != nil {
t.Error(err)
}
wantPos = 2
if pos != wantPos {
t.Errorf("expected pos to be %d, got %d", wantPos, pos)
}
// verify the expected max presence ID
maxPos, err := tab.GetMaxPresenceID(ctx, txn)
if err != nil {
t.Error(err)
}
if maxPos != wantPos {
t.Errorf("expected max pos to be %d, got %d", wantPos, maxPos)
}
// This should increment the position
pos, err = tab.UpsertPresence(ctx, txn, bob.ID, &statusMsg, types.PresenceOnline, timestamp, true)
if err != nil {
t.Error(err)
}
wantPos = pos
if wantPos <= maxPos {
t.Errorf("expected pos to be %d incremented, got %d", wantPos, pos)
}
// This should return only Bobs status
presences, err := tab.GetPresenceAfter(ctx, txn, maxPos, gomatrixserverlib.EventFilter{Limit: 10})
if err != nil {
t.Error(err)
}
if c := len(presences); c > 1 {
t.Errorf("expected only one presence, got %d", c)
}
// Validate the response
wantPresence := &types.PresenceInternal{
UserID: bob.ID,
Presence: types.PresenceOnline,
StreamPos: wantPos,
LastActiveTS: timestamp,
ClientFields: types.PresenceClientResponse{
LastActiveAgo: 0,
Presence: types.PresenceOnline.String(),
StatusMsg: &statusMsg,
},
}
if !reflect.DeepEqual(wantPresence, presences[bob.ID]) {
t.Errorf("unexpected presence result:\n%+v, want\n%+v", presences[bob.ID], wantPresence)
}
// Try getting presences for existing and non-existing users
getUsers := []string{alice.ID, bob.ID, "@doesntexist:test"}
presencesForUsers, err := tab.GetPresenceForUsers(ctx, nil, getUsers)
if err != nil {
t.Error(err)
}
if len(presencesForUsers) >= len(getUsers) {
t.Errorf("expected less presences, but they are the same/more as requested: %d >= %d", len(presencesForUsers), len(getUsers))
}
})
}

80
syncapi/streams/stream_presence.go
View file

@ -17,6 +17,7 @@ package streams
import (
"context"
"encoding/json"
"fmt"
"sync"
"github.com/matrix-org/gomatrixserverlib"
@ -70,39 +71,25 @@ func (p *PresenceStreamProvider) IncrementalSync(
return from
}
if len(presences) == 0 {
getPresenceForUsers, err := p.getNeededUsersFromRequest(ctx, req, presences)
if err != nil {
req.Log.WithError(err).Error("getNeededUsersFromRequest failed")
return from
}
// Got no presence between range and no presence to get from the database
if len(getPresenceForUsers) == 0 && len(presences) == 0 {
return to
}
// add newly joined rooms user presences
newlyJoined := joinedRooms(req.Response, req.Device.UserID)
if len(newlyJoined) > 0 {
// TODO: Check if this is working better than before.
if err = p.notifier.LoadRooms(ctx, p.DB, newlyJoined); err != nil {
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 {
// we already got a presence from this user
if _, ok := presences[roomUsers[i]]; ok {
continue
}
// Bear in mind that this might return nil, but at least populating
// a nil means that there's a map entry so we won't repeat this call.
presences[roomUsers[i]], err = snapshot.GetPresence(ctx, roomUsers[i])
if err != nil {
req.Log.WithError(err).Error("unable to query presence for user")
_ = snapshot.Rollback()
return from
}
if len(presences) > req.Filter.Presence.Limit {
break NewlyJoinedLoop
}
}
}
dbPresences, err := snapshot.GetPresences(ctx, getPresenceForUsers)
if err != nil {
req.Log.WithError(err).Error("unable to query presence for user")
_ = snapshot.Rollback()
return from
}
for _, presence := range dbPresences {
presences[presence.UserID] = presence
}
lastPos := from
@ -164,6 +151,39 @@ func (p *PresenceStreamProvider) IncrementalSync(
return lastPos
}
func (p *PresenceStreamProvider) getNeededUsersFromRequest(ctx context.Context, req *types.SyncRequest, presences map[string]*types.PresenceInternal) ([]string, error) {
getPresenceForUsers := []string{}
// Add presence for users which newly joined a room
for userID := range req.MembershipChanges {
if _, ok := presences[userID]; ok {
continue
}
getPresenceForUsers = append(getPresenceForUsers, userID)
}
// add newly joined rooms user presences
newlyJoined := joinedRooms(req.Response, req.Device.UserID)
if len(newlyJoined) == 0 {
return getPresenceForUsers, nil
}
// TODO: Check if this is working better than before.
if err := p.notifier.LoadRooms(ctx, p.DB, newlyJoined); err != nil {
return getPresenceForUsers, fmt.Errorf("unable to refresh notifier lists: %w", err)
}
for _, roomID := range newlyJoined {
roomUsers := p.notifier.JoinedUsers(roomID)
for i := range roomUsers {
// we already got a presence from this user
if _, ok := presences[roomUsers[i]]; ok {
continue
}
getPresenceForUsers = append(getPresenceForUsers, roomUsers[i])
}
}
return getPresenceForUsers, nil
}
func joinedRooms(res *types.Response, userID string) []string {
var roomIDs []string
for roomID, join := range res.Rooms.Join {

6
syncapi/sync/requestpool.go
View file

@ -145,12 +145,12 @@ func (rp *RequestPool) updatePresence(db storage.Presence, presence string, user
}
// ensure we also send the current status_msg to federated servers and not nil
dbPresence, err := db.GetPresence(context.Background(), userID)
dbPresence, err := db.GetPresences(context.Background(), []string{userID})
if err != nil && err != sql.ErrNoRows {
return
}
if dbPresence != nil {
newPresence.ClientFields = dbPresence.ClientFields
if len(dbPresence) > 0 && dbPresence[0] != nil {
newPresence.ClientFields = dbPresence[0].ClientFields
}
newPresence.ClientFields.Presence = presenceID.String()

4
syncapi/sync/requestpool_test.go
View file

@ -29,8 +29,8 @@ func (d dummyDB) UpdatePresence(ctx context.Context, userID string, presence typ
return 0, nil
}
func (d dummyDB) GetPresence(ctx context.Context, userID string) (*types.PresenceInternal, error) {
return &types.PresenceInternal{}, nil
func (d dummyDB) GetPresences(ctx context.Context, userID []string) ([]*types.PresenceInternal, error) {
return []*types.PresenceInternal{}, nil
}
func (d dummyDB) PresenceAfter(ctx context.Context, after types.StreamPosition, filter gomatrixserverlib.EventFilter) (map[string]*types.PresenceInternal, error) {

2
test/testrig/base.go
View file

@ -108,7 +108,7 @@ func Base(cfg *config.Dendrite) (*base.BaseDendrite, nats.JetStreamContext, *nat
cfg.Global.JetStream.InMemory = true
cfg.SyncAPI.Fulltext.InMemory = true
cfg.FederationAPI.KeyPerspectives = nil
base := base.NewBaseDendrite(cfg, "Tests")
base := base.NewBaseDendrite(cfg, "Tests", base.DisableMetrics)
js, jc := base.NATS.Prepare(base.ProcessContext, &cfg.Global.JetStream)
return base, js, jc
}

71
userapi/inthttp/server.go
View file

@ -16,176 +16,177 @@ package inthttp
import (
"github.com/gorilla/mux"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/userapi/api"
)
// nolint: gocyclo
func AddRoutes(internalAPIMux *mux.Router, s api.UserInternalAPI) {
addRoutesLoginToken(internalAPIMux, s)
func AddRoutes(internalAPIMux *mux.Router, s api.UserInternalAPI, enableMetrics bool) {
addRoutesLoginToken(internalAPIMux, s, enableMetrics)
internalAPIMux.Handle(
PerformAccountCreationPath,
httputil.MakeInternalRPCAPI("UserAPIPerformAccountCreation", s.PerformAccountCreation),
httputil.MakeInternalRPCAPI("UserAPIPerformAccountCreation", enableMetrics, s.PerformAccountCreation),
)
internalAPIMux.Handle(
PerformPasswordUpdatePath,
httputil.MakeInternalRPCAPI("UserAPIPerformPasswordUpdate", s.PerformPasswordUpdate),
httputil.MakeInternalRPCAPI("UserAPIPerformPasswordUpdate", enableMetrics, s.PerformPasswordUpdate),
)
internalAPIMux.Handle(
PerformDeviceCreationPath,
httputil.MakeInternalRPCAPI("UserAPIPerformDeviceCreation", s.PerformDeviceCreation),
httputil.MakeInternalRPCAPI("UserAPIPerformDeviceCreation", enableMetrics, s.PerformDeviceCreation),
)
internalAPIMux.Handle(
PerformLastSeenUpdatePath,
httputil.MakeInternalRPCAPI("UserAPIPerformLastSeenUpdate", s.PerformLastSeenUpdate),
httputil.MakeInternalRPCAPI("UserAPIPerformLastSeenUpdate", enableMetrics, s.PerformLastSeenUpdate),
)
internalAPIMux.Handle(
PerformDeviceUpdatePath,
httputil.MakeInternalRPCAPI("UserAPIPerformDeviceUpdate", s.PerformDeviceUpdate),
httputil.MakeInternalRPCAPI("UserAPIPerformDeviceUpdate", enableMetrics, s.PerformDeviceUpdate),
)
internalAPIMux.Handle(
PerformDeviceDeletionPath,
httputil.MakeInternalRPCAPI("UserAPIPerformDeviceDeletion", s.PerformDeviceDeletion),
httputil.MakeInternalRPCAPI("UserAPIPerformDeviceDeletion", enableMetrics, s.PerformDeviceDeletion),
)
internalAPIMux.Handle(
PerformAccountDeactivationPath,
httputil.MakeInternalRPCAPI("UserAPIPerformAccountDeactivation", s.PerformAccountDeactivation),
httputil.MakeInternalRPCAPI("UserAPIPerformAccountDeactivation", enableMetrics, s.PerformAccountDeactivation),
)
internalAPIMux.Handle(
PerformOpenIDTokenCreationPath,
httputil.MakeInternalRPCAPI("UserAPIPerformOpenIDTokenCreation", s.PerformOpenIDTokenCreation),
httputil.MakeInternalRPCAPI("UserAPIPerformOpenIDTokenCreation", enableMetrics, s.PerformOpenIDTokenCreation),
)
internalAPIMux.Handle(
QueryProfilePath,
httputil.MakeInternalRPCAPI("UserAPIQueryProfile", s.QueryProfile),
httputil.MakeInternalRPCAPI("UserAPIQueryProfile", enableMetrics, s.QueryProfile),
)
internalAPIMux.Handle(
QueryAccessTokenPath,
httputil.MakeInternalRPCAPI("UserAPIQueryAccessToken", s.QueryAccessToken),
httputil.MakeInternalRPCAPI("UserAPIQueryAccessToken", enableMetrics, s.QueryAccessToken),
)
internalAPIMux.Handle(
QueryDevicesPath,
httputil.MakeInternalRPCAPI("UserAPIQueryDevices", s.QueryDevices),
httputil.MakeInternalRPCAPI("UserAPIQueryDevices", enableMetrics, s.QueryDevices),
)
internalAPIMux.Handle(
QueryAccountDataPath,
httputil.MakeInternalRPCAPI("UserAPIQueryAccountData", s.QueryAccountData),
httputil.MakeInternalRPCAPI("UserAPIQueryAccountData", enableMetrics, s.QueryAccountData),
)
internalAPIMux.Handle(
QueryDeviceInfosPath,
httputil.MakeInternalRPCAPI("UserAPIQueryDeviceInfos", s.QueryDeviceInfos),
httputil.MakeInternalRPCAPI("UserAPIQueryDeviceInfos", enableMetrics, s.QueryDeviceInfos),
)
internalAPIMux.Handle(
QuerySearchProfilesPath,
httputil.MakeInternalRPCAPI("UserAPIQuerySearchProfiles", s.QuerySearchProfiles),
httputil.MakeInternalRPCAPI("UserAPIQuerySearchProfiles", enableMetrics, s.QuerySearchProfiles),
)
internalAPIMux.Handle(
QueryOpenIDTokenPath,
httputil.MakeInternalRPCAPI("UserAPIQueryOpenIDToken", s.QueryOpenIDToken),
httputil.MakeInternalRPCAPI("UserAPIQueryOpenIDToken", enableMetrics, s.QueryOpenIDToken),
)
internalAPIMux.Handle(
InputAccountDataPath,
httputil.MakeInternalRPCAPI("UserAPIInputAccountData", s.InputAccountData),
httputil.MakeInternalRPCAPI("UserAPIInputAccountData", enableMetrics, s.InputAccountData),
)
internalAPIMux.Handle(
QueryKeyBackupPath,
httputil.MakeInternalRPCAPI("UserAPIQueryKeyBackup", s.QueryKeyBackup),
httputil.MakeInternalRPCAPI("UserAPIQueryKeyBackup", enableMetrics, s.QueryKeyBackup),
)
internalAPIMux.Handle(
PerformKeyBackupPath,
httputil.MakeInternalRPCAPI("UserAPIPerformKeyBackup", s.PerformKeyBackup),
httputil.MakeInternalRPCAPI("UserAPIPerformKeyBackup", enableMetrics, s.PerformKeyBackup),
)
internalAPIMux.Handle(
QueryNotificationsPath,
httputil.MakeInternalRPCAPI("UserAPIQueryNotifications", s.QueryNotifications),
httputil.MakeInternalRPCAPI("UserAPIQueryNotifications", enableMetrics, s.QueryNotifications),
)
internalAPIMux.Handle(
PerformPusherSetPath,
httputil.MakeInternalRPCAPI("UserAPIPerformPusherSet", s.PerformPusherSet),
httputil.MakeInternalRPCAPI("UserAPIPerformPusherSet", enableMetrics, s.PerformPusherSet),
)
internalAPIMux.Handle(
PerformPusherDeletionPath,
httputil.MakeInternalRPCAPI("UserAPIPerformPusherDeletion", s.PerformPusherDeletion),
httputil.MakeInternalRPCAPI("UserAPIPerformPusherDeletion", enableMetrics, s.PerformPusherDeletion),
)
internalAPIMux.Handle(
QueryPushersPath,
httputil.MakeInternalRPCAPI("UserAPIQueryPushers", s.QueryPushers),
httputil.MakeInternalRPCAPI("UserAPIQueryPushers", enableMetrics, s.QueryPushers),
)
internalAPIMux.Handle(
PerformPushRulesPutPath,
httputil.MakeInternalRPCAPI("UserAPIPerformPushRulesPut", s.PerformPushRulesPut),
httputil.MakeInternalRPCAPI("UserAPIPerformPushRulesPut", enableMetrics, s.PerformPushRulesPut),
)
internalAPIMux.Handle(
QueryPushRulesPath,
httputil.MakeInternalRPCAPI("UserAPIQueryPushRules", s.QueryPushRules),
httputil.MakeInternalRPCAPI("UserAPIQueryPushRules", enableMetrics, s.QueryPushRules),
)
internalAPIMux.Handle(
PerformSetAvatarURLPath,
httputil.MakeInternalRPCAPI("UserAPIPerformSetAvatarURL", s.SetAvatarURL),
httputil.MakeInternalRPCAPI("UserAPIPerformSetAvatarURL", enableMetrics, s.SetAvatarURL),
)
internalAPIMux.Handle(
QueryNumericLocalpartPath,
httputil.MakeInternalRPCAPI("UserAPIQueryNumericLocalpart", s.QueryNumericLocalpart),
httputil.MakeInternalRPCAPI("UserAPIQueryNumericLocalpart", enableMetrics, s.QueryNumericLocalpart),
)
internalAPIMux.Handle(
QueryAccountAvailabilityPath,
httputil.MakeInternalRPCAPI("UserAPIQueryAccountAvailability", s.QueryAccountAvailability),
httputil.MakeInternalRPCAPI("UserAPIQueryAccountAvailability", enableMetrics, s.QueryAccountAvailability),
)
internalAPIMux.Handle(
QueryAccountByPasswordPath,
httputil.MakeInternalRPCAPI("UserAPIQueryAccountByPassword", s.QueryAccountByPassword),
httputil.MakeInternalRPCAPI("UserAPIQueryAccountByPassword", enableMetrics, s.QueryAccountByPassword),
)
internalAPIMux.Handle(
PerformSetDisplayNamePath,
httputil.MakeInternalRPCAPI("UserAPISetDisplayName", s.SetDisplayName),
httputil.MakeInternalRPCAPI("UserAPISetDisplayName", enableMetrics, s.SetDisplayName),
)
internalAPIMux.Handle(
QueryLocalpartForThreePIDPath,
httputil.MakeInternalRPCAPI("UserAPIQueryLocalpartForThreePID", s.QueryLocalpartForThreePID),
httputil.MakeInternalRPCAPI("UserAPIQueryLocalpartForThreePID", enableMetrics, s.QueryLocalpartForThreePID),
)
internalAPIMux.Handle(
QueryThreePIDsForLocalpartPath,
httputil.MakeInternalRPCAPI("UserAPIQueryThreePIDsForLocalpart", s.QueryThreePIDsForLocalpart),
httputil.MakeInternalRPCAPI("UserAPIQueryThreePIDsForLocalpart", enableMetrics, s.QueryThreePIDsForLocalpart),
)
internalAPIMux.Handle(
PerformForgetThreePIDPath,
httputil.MakeInternalRPCAPI("UserAPIPerformForgetThreePID", s.PerformForgetThreePID),
httputil.MakeInternalRPCAPI("UserAPIPerformForgetThreePID", enableMetrics, s.PerformForgetThreePID),
)
internalAPIMux.Handle(
PerformSaveThreePIDAssociationPath,
httputil.MakeInternalRPCAPI("UserAPIPerformSaveThreePIDAssociation", s.PerformSaveThreePIDAssociation),
httputil.MakeInternalRPCAPI("UserAPIPerformSaveThreePIDAssociation", enableMetrics, s.PerformSaveThreePIDAssociation),
)
}

9
userapi/inthttp/server_logintoken.go
View file

@ -16,24 +16,25 @@ package inthttp
import (
"github.com/gorilla/mux"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/userapi/api"
)
// addRoutesLoginToken adds routes for all login token API calls.
func addRoutesLoginToken(internalAPIMux *mux.Router, s api.UserInternalAPI) {
func addRoutesLoginToken(internalAPIMux *mux.Router, s api.UserInternalAPI, enableMetrics bool) {
internalAPIMux.Handle(
PerformLoginTokenCreationPath,
httputil.MakeInternalRPCAPI("UserAPIPerformLoginTokenCreation", s.PerformLoginTokenCreation),
httputil.MakeInternalRPCAPI("UserAPIPerformLoginTokenCreation", enableMetrics, s.PerformLoginTokenCreation),
)
internalAPIMux.Handle(
PerformLoginTokenDeletionPath,
httputil.MakeInternalRPCAPI("UserAPIPerformLoginTokenDeletion", s.PerformLoginTokenDeletion),
httputil.MakeInternalRPCAPI("UserAPIPerformLoginTokenDeletion", enableMetrics, s.PerformLoginTokenDeletion),
)
internalAPIMux.Handle(
QueryLoginTokenPath,
httputil.MakeInternalRPCAPI("UserAPIQueryLoginToken", s.QueryLoginToken),
httputil.MakeInternalRPCAPI("UserAPIQueryLoginToken", enableMetrics, s.QueryLoginToken),
)
}

4
userapi/userapi.go
View file

@ -37,8 +37,8 @@ import (
// AddInternalRoutes registers HTTP handlers for the internal API. Invokes functions
// on the given input API.
func AddInternalRoutes(router *mux.Router, intAPI api.UserInternalAPI) {
inthttp.AddRoutes(router, intAPI)
func AddInternalRoutes(router *mux.Router, intAPI api.UserInternalAPI, enableMetrics bool) {
inthttp.AddRoutes(router, intAPI, enableMetrics)
}
// NewInternalAPI returns a concerete implementation of the internal API. Callers

55
userapi/userapi_test.go
View file

@ -27,14 +27,13 @@ import (
"golang.org/x/crypto/bcrypt"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/test"
"github.com/matrix-org/dendrite/test/testrig"
"github.com/matrix-org/dendrite/userapi"
"github.com/matrix-org/dendrite/userapi/inthttp"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/dendrite/userapi/internal"
"github.com/matrix-org/dendrite/userapi/inthttp"
"github.com/matrix-org/dendrite/userapi/storage"
)
@ -79,19 +78,6 @@ func MustMakeInternalAPI(t *testing.T, opts apiTestOpts, dbType test.DBType) (ap
func TestQueryProfile(t *testing.T) {
aliceAvatarURL := "mxc://example.com/alice"
aliceDisplayName := "Alice"
// only one DBType, since userapi.AddInternalRoutes complains about multiple prometheus counters added
userAPI, accountDB, close := MustMakeInternalAPI(t, apiTestOpts{}, test.DBTypeSQLite)
defer close()
_, err := accountDB.CreateAccount(context.TODO(), "alice", serverName, "foobar", "", api.AccountTypeUser)
if err != nil {
t.Fatalf("failed to make account: %s", err)
}
if _, _, err := accountDB.SetAvatarURL(context.TODO(), "alice", serverName, aliceAvatarURL); err != nil {
t.Fatalf("failed to set avatar url: %s", err)
}
if _, _, err := accountDB.SetDisplayName(context.TODO(), "alice", serverName, aliceDisplayName); err != nil {
t.Fatalf("failed to set display name: %s", err)
}
testCases := []struct {
req api.QueryProfileRequest
@ -142,19 +128,34 @@ func TestQueryProfile(t *testing.T) {
}
}
t.Run("HTTP API", func(t *testing.T) {
router := mux.NewRouter().PathPrefix(httputil.InternalPathPrefix).Subrouter()
userapi.AddInternalRoutes(router, userAPI)
apiURL, cancel := test.ListenAndServe(t, router, false)
defer cancel()
httpAPI, err := inthttp.NewUserAPIClient(apiURL, &http.Client{})
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
userAPI, accountDB, close := MustMakeInternalAPI(t, apiTestOpts{}, dbType)
defer close()
_, err := accountDB.CreateAccount(context.TODO(), "alice", serverName, "foobar", "", api.AccountTypeUser)
if err != nil {
t.Fatalf("failed to create HTTP client")
t.Fatalf("failed to make account: %s", err)
}
runCases(httpAPI, true)
})
t.Run("Monolith", func(t *testing.T) {
runCases(userAPI, false)
if _, _, err := accountDB.SetAvatarURL(context.TODO(), "alice", serverName, aliceAvatarURL); err != nil {
t.Fatalf("failed to set avatar url: %s", err)
}
if _, _, err := accountDB.SetDisplayName(context.TODO(), "alice", serverName, aliceDisplayName); err != nil {
t.Fatalf("failed to set display name: %s", err)
}
t.Run("HTTP API", func(t *testing.T) {
router := mux.NewRouter().PathPrefix(httputil.InternalPathPrefix).Subrouter()
userapi.AddInternalRoutes(router, userAPI, false)
apiURL, cancel := test.ListenAndServe(t, router, false)
defer cancel()
httpAPI, err := inthttp.NewUserAPIClient(apiURL, &http.Client{})
if err != nil {
t.Fatalf("failed to create HTTP client")
}
runCases(httpAPI, true)
})
t.Run("Monolith", func(t *testing.T) {
runCases(userAPI, false)
})
})
}

119
userapi/util/notify_test.go Normal file
View file

@ -0,0 +1,119 @@
package util_test
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"time"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
"golang.org/x/crypto/bcrypt"
"github.com/matrix-org/dendrite/internal/pushgateway"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/test"
"github.com/matrix-org/dendrite/test/testrig"
"github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/dendrite/userapi/storage"
userUtil "github.com/matrix-org/dendrite/userapi/util"
)
func TestNotifyUserCountsAsync(t *testing.T) {
alice := test.NewUser(t)
aliceLocalpart, serverName, err := gomatrixserverlib.SplitID('@', alice.ID)
if err != nil {
t.Error(err)
}
ctx := context.Background()
// Create a test room, just used to provide events
room := test.NewRoom(t, alice)
dummyEvent := room.Events()[len(room.Events())-1]
appID := util.RandomString(8)
pushKey := util.RandomString(8)
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
receivedRequest := make(chan bool, 1)
// create a test server which responds to our /notify call
srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
var data pushgateway.NotifyRequest
if err := json.NewDecoder(r.Body).Decode(&data); err != nil {
t.Error(err)
}
notification := data.Notification
// Validate the request
if notification.Counts == nil {
t.Fatal("no unread notification counts in request")
}
if unread := notification.Counts.Unread; unread != 1 {
t.Errorf("expected one unread notification, got %d", unread)
}
if len(notification.Devices) == 0 {
t.Fatal("expected devices in request")
}
// We only created one push device, so access it directly
device := notification.Devices[0]
if device.AppID != appID {
t.Errorf("unexpected app_id: %s, want %s", device.AppID, appID)
}
if device.PushKey != pushKey {
t.Errorf("unexpected push_key: %s, want %s", device.PushKey, pushKey)
}
// Return empty result, otherwise the call is handled as failed
if _, err := w.Write([]byte("{}")); err != nil {
t.Error(err)
}
close(receivedRequest)
}))
defer srv.Close()
// Create DB and Dendrite base
connStr, close := test.PrepareDBConnectionString(t, dbType)
defer close()
base, _, _ := testrig.Base(nil)
defer base.Close()
db, err := storage.NewUserAPIDatabase(base, &config.DatabaseOptions{
ConnectionString: config.DataSource(connStr),
}, "test", bcrypt.MinCost, 0, 0, "")
if err != nil {
t.Error(err)
}
// Prepare pusher with our test server URL
if err := db.UpsertPusher(ctx, api.Pusher{
Kind: api.HTTPKind,
AppID: appID,
PushKey: pushKey,
Data: map[string]interface{}{
"url": srv.URL,
},
}, aliceLocalpart, serverName); err != nil {
t.Error(err)
}
// Insert a dummy event
if err := db.InsertNotification(ctx, aliceLocalpart, serverName, dummyEvent.EventID(), 0, nil, &api.Notification{
Event: gomatrixserverlib.HeaderedToClientEvent(dummyEvent, gomatrixserverlib.FormatAll),
}); err != nil {
t.Error(err)
}
// Notify the user about a new notification
if err := userUtil.NotifyUserCountsAsync(ctx, pushgateway.NewHTTPClient(true), aliceLocalpart, serverName, db); err != nil {
t.Error(err)
}
select {
case <-time.After(time.Second * 5):
t.Error("timed out waiting for response")
case <-receivedRequest:
}
})
}

10
userapi/util/phonehomestats.go
View file

@ -97,12 +97,10 @@ func (p *phoneHomeStats) collect() {
// configuration information
p.stats["federation_disabled"] = p.cfg.Global.DisableFederation
p.stats["nats_embedded"] = true
p.stats["nats_in_memory"] = p.cfg.Global.JetStream.InMemory
if len(p.cfg.Global.JetStream.Addresses) > 0 {
p.stats["nats_embedded"] = false
p.stats["nats_in_memory"] = false // probably
}
natsEmbedded := len(p.cfg.Global.JetStream.Addresses) == 0
p.stats["nats_embedded"] = natsEmbedded
p.stats["nats_in_memory"] = p.cfg.Global.JetStream.InMemory && natsEmbedded
if len(p.cfg.Logging) > 0 {
p.stats["log_level"] = p.cfg.Logging[0].Level
} else {

84
userapi/util/phonehomestats_test.go Normal file
View file

@ -0,0 +1,84 @@
package util
import (
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"time"
"golang.org/x/crypto/bcrypt"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/test"
"github.com/matrix-org/dendrite/test/testrig"
"github.com/matrix-org/dendrite/userapi/storage"
)
func TestCollect(t *testing.T) {
test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
b, _, _ := testrig.Base(nil)
connStr, closeDB := test.PrepareDBConnectionString(t, dbType)
defer closeDB()
db, err := storage.NewUserAPIDatabase(b, &config.DatabaseOptions{
ConnectionString: config.DataSource(connStr),
}, "localhost", bcrypt.MinCost, 1000, 1000, "")
if err != nil {
t.Error(err)
}
receivedRequest := make(chan struct{}, 1)
// create a test server which responds to our call
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
var data map[string]interface{}
if err := json.NewDecoder(r.Body).Decode(&data); err != nil {
t.Error(err)
}
defer r.Body.Close()
if _, err := w.Write([]byte("{}")); err != nil {
t.Error(err)
}
// verify the received data matches our expectations
dbEngine, ok := data["database_engine"]
if !ok {
t.Errorf("missing database_engine in JSON request: %+v", data)
}
version, ok := data["version"]
if !ok {
t.Errorf("missing version in JSON request: %+v", data)
}
if version != internal.VersionString() {
t.Errorf("unexpected version: %q, expected %q", version, internal.VersionString())
}
switch {
case dbType == test.DBTypeSQLite && dbEngine != "SQLite":
t.Errorf("unexpected database_engine: %s", dbEngine)
case dbType == test.DBTypePostgres && dbEngine != "Postgres":
t.Errorf("unexpected database_engine: %s", dbEngine)
}
close(receivedRequest)
}))
defer srv.Close()
b.Cfg.Global.ReportStats.Endpoint = srv.URL
stats := phoneHomeStats{
prevData: timestampToRUUsage{},
serverName: "localhost",
startTime: time.Now(),
cfg: b.Cfg,
db: db,
isMonolith: false,
client: &http.Client{Timeout: time.Second},
}
stats.collect()
select {
case <-time.After(time.Second * 5):
t.Error("timed out waiting for response")
case <-receivedRequest:
}
})
}