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

BaseDendrite be gone!

Browse source
This commit is contained in:
Till Faelligen 2023-03-21 13:06:17 +01:00
parent 85f76d7491
commit b5789f24c0
No known key found for this signature in database
GPG key ID: ACCDC9606D472758
25 changed files with 468 additions and 498 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -31,11 +31,12 @@ import (
"github.com/matrix-org/dendrite/federationapi"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver"
"github.com/matrix-org/dendrite/setup"
"github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/setup/process"
"github.com/matrix-org/dendrite/userapi"
"github.com/matrix-org/gomatrixserverlib"
@ -159,9 +160,8 @@ func startup() {
pManager.AddPeer("wss://pinecone.matrix.org/public")
cfg := &config.Dendrite{}
cfg.Defaults(true)
cfg.Defaults(config.DefaultOpts{Generate: true, SingleDatabase: false})
cfg.UserAPI.AccountDatabase.ConnectionString = "file:/idb/dendritejs_account.db"
cfg.AppServiceAPI.Database.ConnectionString = "file:/idb/dendritejs_appservice.db"
cfg.FederationAPI.Database.ConnectionString = "file:/idb/dendritejs_fedsender.db"
cfg.MediaAPI.Database.ConnectionString = "file:/idb/dendritejs_mediaapi.db"
cfg.RoomServer.Database.ConnectionString = "file:/idb/dendritejs_roomserver.db"
@ -178,30 +178,30 @@ func startup() {
if err := cfg.Derive(); err != nil {
logrus.Fatalf("Failed to derive values from config: %s", err)
}
base := base.NewBaseDendrite(cfg)
defer base.Close() // nolint: errcheck
natsInstance := jetstream.NATSInstance{}
processCtx := process.NewProcessContext()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
routers := httputil.NewRouters()
caches := caching.NewRistrettoCache(cfg.Global.Cache.EstimatedMaxSize, cfg.Global.Cache.MaxAge, caching.EnableMetrics)
rsAPI := roomserver.NewInternalAPI(base.ProcessContext, base.Cfg, base.ConnectionManager, &natsInstance, caches, base.EnableMetrics)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.EnableMetrics)
federation := conn.CreateFederationClient(base, pSessions)
federation := conn.CreateFederationClient(cfg, pSessions)
serverKeyAPI := &signing.YggdrasilKeys{}
keyRing := serverKeyAPI.KeyRing()
userAPI := userapi.NewInternalAPI(base.ProcessContext, base.Cfg, base.ConnectionManager, &natsInstance, rsAPI, federation)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, federation)
asQuery := appservice.NewInternalAPI(
base.ProcessContext, base.Cfg, &natsInstance, userAPI, rsAPI,
processCtx, cfg, &natsInstance, userAPI, rsAPI,
)
rsAPI.SetAppserviceAPI(asQuery)
caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.EnableMetrics)
fedSenderAPI := federationapi.NewInternalAPI(base.ProcessContext, base.Cfg, base.ConnectionManager, &natsInstance, federation, rsAPI, caches, keyRing, true)
fedSenderAPI := federationapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, federation, rsAPI, caches, keyRing, true)
rsAPI.SetFederationAPI(fedSenderAPI, keyRing)
monolith := setup.Monolith{
Config: base.Cfg,
Client: conn.CreateClient(base, pSessions),
Config: cfg,
Client: conn.CreateClient(pSessions),
FedClient: federation,
KeyRing: keyRing,
@ -212,15 +212,15 @@ func startup() {
//ServerKeyAPI: serverKeyAPI,
ExtPublicRoomsProvider: rooms.NewPineconeRoomProvider(pRouter, pSessions, fedSenderAPI, federation),
}
monolith.AddAllPublicRoutes(base, &natsInstance, caches)
monolith.AddAllPublicRoutes(processCtx, cfg, routers, cm, &natsInstance, caches, caching.EnableMetrics)
httpRouter := mux.NewRouter().SkipClean(true).UseEncodedPath()
httpRouter.PathPrefix(httputil.PublicClientPathPrefix).Handler(base.Routers.Client)
httpRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(base.Routers.Media)
httpRouter.PathPrefix(httputil.PublicClientPathPrefix).Handler(routers.Client)
httpRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(routers.Media)
p2pRouter := pSessions.Protocol("matrix").HTTP().Mux()
p2pRouter.Handle(httputil.PublicFederationPathPrefix, base.Routers.Federation)
p2pRouter.Handle(httputil.PublicMediaPathPrefix, base.Routers.Media)
p2pRouter.Handle(httputil.PublicFederationPathPrefix, routers.Federation)
p2pRouter.Handle(httputil.PublicMediaPathPrefix, routers.Media)
// Expose the matrix APIs via fetch - for local traffic
go func() {

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

@ -30,6 +30,9 @@ import (
"github.com/matrix-org/dendrite/cmd/dendrite-demo-pinecone/relay"
"github.com/matrix-org/dendrite/cmd/dendrite-demo-yggdrasil/signing"
"github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/setup/process"
userapiAPI "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/pinecone/types"
@ -187,7 +190,7 @@ func (m *DendriteMonolith) SetRelayServers(nodeID string, uris string) {
relay.UpdateNodeRelayServers(
gomatrixserverlib.ServerName(nodeKey),
relays,
m.p2pMonolith.BaseDendrite.Context(),
m.p2pMonolith.ProcessCtx.Context(),
m.p2pMonolith.GetFederationAPI(),
)
}
@ -214,7 +217,7 @@ func (m *DendriteMonolith) GetRelayServers(nodeID string) string {
} else {
request := api.P2PQueryRelayServersRequest{Server: gomatrixserverlib.ServerName(nodeKey)}
response := api.P2PQueryRelayServersResponse{}
err := m.p2pMonolith.GetFederationAPI().P2PQueryRelayServers(m.p2pMonolith.BaseDendrite.Context(), &request, &response)
err := m.p2pMonolith.GetFederationAPI().P2PQueryRelayServers(m.p2pMonolith.ProcessCtx.Context(), &request, &response)
if err != nil {
logrus.Warnf("Failed obtaining list of this node's relay servers: %s", err.Error())
return ""
@ -346,10 +349,14 @@ func (m *DendriteMonolith) Start() {
// This isn't actually fixed: https://github.com/blevesearch/zapx/pull/147
cfg.SyncAPI.Fulltext.Enabled = false
processCtx := process.NewProcessContext()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
routers := httputil.NewRouters()
enableRelaying := false
enableMetrics := false
enableWebsockets := false
m.p2pMonolith.SetupDendrite(cfg, 65432, enableRelaying, enableMetrics, enableWebsockets)
m.p2pMonolith.SetupDendrite(processCtx, cfg, cm, routers, 65432, enableRelaying, enableMetrics, enableWebsockets)
m.p2pMonolith.StartMonolith()
}

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

@ -12,6 +12,7 @@ import (
"path/filepath"
"time"
"github.com/getsentry/sentry-go"
"github.com/gorilla/mux"
"github.com/matrix-org/dendrite/appservice"
"github.com/matrix-org/dendrite/cmd/dendrite-demo-yggdrasil/signing"
@ -19,11 +20,13 @@ import (
"github.com/matrix-org/dendrite/cmd/dendrite-demo-yggdrasil/yggrooms"
"github.com/matrix-org/dendrite/federationapi"
"github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver"
"github.com/matrix-org/dendrite/setup"
"github.com/matrix-org/dendrite/setup/base"
basepkg "github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/setup/process"
@ -150,27 +153,71 @@ func (m *DendriteMonolith) Start() {
panic(err)
}
base := base.NewBaseDendrite(cfg)
base.ConfigureAdminEndpoints()
m.processContext = base.ProcessContext
defer base.Close() // nolint: errcheck
configErrors := &config.ConfigErrors{}
cfg.Verify(configErrors)
if len(*configErrors) > 0 {
for _, err := range *configErrors {
logrus.Errorf("Configuration error: %s", err)
}
logrus.Fatalf("Failed to start due to configuration errors")
}
federation := ygg.CreateFederationClient(base)
internal.SetupStdLogging()
internal.SetupHookLogging(cfg.Logging)
internal.SetupPprof()
logrus.Infof("Dendrite version %s", internal.VersionString())
if !cfg.ClientAPI.RegistrationDisabled && cfg.ClientAPI.OpenRegistrationWithoutVerificationEnabled {
logrus.Warn("Open registration is enabled")
}
closer, err := cfg.SetupTracing()
if err != nil {
logrus.WithError(err).Panicf("failed to start opentracing")
}
defer closer.Close()
if cfg.Global.Sentry.Enabled {
logrus.Info("Setting up Sentry for debugging...")
err = sentry.Init(sentry.ClientOptions{
Dsn: cfg.Global.Sentry.DSN,
Environment: cfg.Global.Sentry.Environment,
Debug: true,
ServerName: string(cfg.Global.ServerName),
Release: "dendrite@" + internal.VersionString(),
AttachStacktrace: true,
})
if err != nil {
logrus.WithError(err).Panic("failed to start Sentry")
}
}
processCtx := process.NewProcessContext()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
routers := httputil.NewRouters()
basepkg.ConfigureAdminEndpoints(processCtx, routers)
m.processContext = processCtx
defer func() {
processCtx.ShutdownDendrite()
processCtx.WaitForShutdown()
}() // nolint: errcheck
federation := ygg.CreateFederationClient(cfg)
serverKeyAPI := &signing.YggdrasilKeys{}
keyRing := serverKeyAPI.KeyRing()
caches := caching.NewRistrettoCache(cfg.Global.Cache.EstimatedMaxSize, cfg.Global.Cache.MaxAge, caching.DisableMetrics)
caches := caching.NewRistrettoCache(cfg.Global.Cache.EstimatedMaxSize, cfg.Global.Cache.MaxAge, caching.EnableMetrics)
natsInstance := jetstream.NATSInstance{}
rsAPI := roomserver.NewInternalAPI(base.ProcessContext, base.Cfg, base.ConnectionManager, &natsInstance, caches, base.EnableMetrics)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.EnableMetrics)
fsAPI := federationapi.NewInternalAPI(
base.ProcessContext, base.Cfg, base.ConnectionManager, &natsInstance, federation, rsAPI, caches, keyRing, true,
processCtx, cfg, cm, &natsInstance, federation, rsAPI, caches, keyRing, true,
)
userAPI := userapi.NewInternalAPI(base.ProcessContext, base.Cfg, base.ConnectionManager, &natsInstance, rsAPI, federation)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, federation)
asAPI := appservice.NewInternalAPI(base.ProcessContext, base.Cfg, &natsInstance, userAPI, rsAPI)
asAPI := appservice.NewInternalAPI(processCtx, cfg, &natsInstance, userAPI, rsAPI)
rsAPI.SetAppserviceAPI(asAPI)
// The underlying roomserver implementation needs to be able to call the fedsender.
@ -178,8 +225,8 @@ func (m *DendriteMonolith) Start() {
rsAPI.SetFederationAPI(fsAPI, keyRing)
monolith := setup.Monolith{
Config: base.Cfg,
Client: ygg.CreateClient(base),
Config: cfg,
Client: ygg.CreateClient(),
FedClient: federation,
KeyRing: keyRing,
@ -191,17 +238,17 @@ func (m *DendriteMonolith) Start() {
ygg, fsAPI, federation,
),
}
monolith.AddAllPublicRoutes(base, &natsInstance, caches)
monolith.AddAllPublicRoutes(processCtx, cfg, routers, cm, &natsInstance, caches, caching.EnableMetrics)
httpRouter := mux.NewRouter()
httpRouter.PathPrefix(httputil.PublicClientPathPrefix).Handler(base.Routers.Client)
httpRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(base.Routers.Media)
httpRouter.PathPrefix(httputil.DendriteAdminPathPrefix).Handler(base.Routers.DendriteAdmin)
httpRouter.PathPrefix(httputil.SynapseAdminPathPrefix).Handler(base.Routers.SynapseAdmin)
httpRouter.PathPrefix(httputil.PublicClientPathPrefix).Handler(routers.Client)
httpRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(routers.Media)
httpRouter.PathPrefix(httputil.DendriteAdminPathPrefix).Handler(routers.DendriteAdmin)
httpRouter.PathPrefix(httputil.SynapseAdminPathPrefix).Handler(routers.SynapseAdmin)
yggRouter := mux.NewRouter()
yggRouter.PathPrefix(httputil.PublicFederationPathPrefix).Handler(base.Routers.Federation)
yggRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(base.Routers.Media)
yggRouter.PathPrefix(httputil.PublicFederationPathPrefix).Handler(routers.Federation)
yggRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(routers.Media)
// Build both ends of a HTTP multiplex.
m.httpServer = &http.Server{

46
clientapi/routing/auth_fallback_test.go
View file

@ -10,30 +10,28 @@ import (
"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/test/testrig"
)
func Test_AuthFallback(t *testing.T) {
base, _, _ := testrig.Base(nil)
defer base.Close()
cfg := config.Dendrite{}
cfg.Defaults(config.DefaultOpts{Generate: true, SingleDatabase: true})
for _, useHCaptcha := range []bool{false, true} {
for _, recaptchaEnabled := range []bool{false, true} {
for _, wantErr := range []bool{false, true} {
t.Run(fmt.Sprintf("useHCaptcha(%v) - recaptchaEnabled(%v) - wantErr(%v)", useHCaptcha, recaptchaEnabled, wantErr), func(t *testing.T) {
// Set the defaults for each test
base.Cfg.ClientAPI.Defaults(config.DefaultOpts{Generate: true, SingleDatabase: true})
base.Cfg.ClientAPI.RecaptchaEnabled = recaptchaEnabled
base.Cfg.ClientAPI.RecaptchaPublicKey = "pub"
base.Cfg.ClientAPI.RecaptchaPrivateKey = "priv"
cfg.ClientAPI.Defaults(config.DefaultOpts{Generate: true, SingleDatabase: true})
cfg.ClientAPI.RecaptchaEnabled = recaptchaEnabled
cfg.ClientAPI.RecaptchaPublicKey = "pub"
cfg.ClientAPI.RecaptchaPrivateKey = "priv"
if useHCaptcha {
base.Cfg.ClientAPI.RecaptchaSiteVerifyAPI = "https://hcaptcha.com/siteverify"
base.Cfg.ClientAPI.RecaptchaApiJsUrl = "https://js.hcaptcha.com/1/api.js"
base.Cfg.ClientAPI.RecaptchaFormField = "h-captcha-response"
base.Cfg.ClientAPI.RecaptchaSitekeyClass = "h-captcha"
cfg.ClientAPI.RecaptchaSiteVerifyAPI = "https://hcaptcha.com/siteverify"
cfg.ClientAPI.RecaptchaApiJsUrl = "https://js.hcaptcha.com/1/api.js"
cfg.ClientAPI.RecaptchaFormField = "h-captcha-response"
cfg.ClientAPI.RecaptchaSitekeyClass = "h-captcha"
}
cfgErrs := &config.ConfigErrors{}
base.Cfg.ClientAPI.Verify(cfgErrs)
cfg.ClientAPI.Verify(cfgErrs)
if len(*cfgErrs) > 0 {
t.Fatalf("(hCaptcha=%v) unexpected config errors: %s", useHCaptcha, cfgErrs.Error())
}
@ -41,7 +39,7 @@ func Test_AuthFallback(t *testing.T) {
req := httptest.NewRequest(http.MethodGet, "/?session=1337", nil)
rec := httptest.NewRecorder()
AuthFallback(rec, req, authtypes.LoginTypeRecaptcha, &base.Cfg.ClientAPI)
AuthFallback(rec, req, authtypes.LoginTypeRecaptcha, &cfg.ClientAPI)
if !recaptchaEnabled {
if rec.Code != http.StatusBadRequest {
t.Fatalf("unexpected response code: %d, want %d", rec.Code, http.StatusBadRequest)
@ -50,8 +48,8 @@ func Test_AuthFallback(t *testing.T) {
t.Fatalf("unexpected response body: %s", rec.Body.String())
}
} else {
if !strings.Contains(rec.Body.String(), base.Cfg.ClientAPI.RecaptchaSitekeyClass) {
t.Fatalf("body does not contain %s: %s", base.Cfg.ClientAPI.RecaptchaSitekeyClass, rec.Body.String())
if !strings.Contains(rec.Body.String(), cfg.ClientAPI.RecaptchaSitekeyClass) {
t.Fatalf("body does not contain %s: %s", cfg.ClientAPI.RecaptchaSitekeyClass, rec.Body.String())
}
}
@ -64,14 +62,14 @@ func Test_AuthFallback(t *testing.T) {
}))
defer srv.Close() // nolint: errcheck
base.Cfg.ClientAPI.RecaptchaSiteVerifyAPI = srv.URL
cfg.ClientAPI.RecaptchaSiteVerifyAPI = srv.URL
// check the result after sending the captcha
req = httptest.NewRequest(http.MethodPost, "/?session=1337", nil)
req.Form = url.Values{}
req.Form.Add(base.Cfg.ClientAPI.RecaptchaFormField, "someRandomValue")
req.Form.Add(cfg.ClientAPI.RecaptchaFormField, "someRandomValue")
rec = httptest.NewRecorder()
AuthFallback(rec, req, authtypes.LoginTypeRecaptcha, &base.Cfg.ClientAPI)
AuthFallback(rec, req, authtypes.LoginTypeRecaptcha, &cfg.ClientAPI)
if recaptchaEnabled {
if !wantErr {
if rec.Code != http.StatusOK {
@ -105,7 +103,7 @@ func Test_AuthFallback(t *testing.T) {
t.Run("unknown fallbacks are handled correctly", func(t *testing.T) {
req := httptest.NewRequest(http.MethodPost, "/?session=1337", nil)
rec := httptest.NewRecorder()
AuthFallback(rec, req, "DoesNotExist", &base.Cfg.ClientAPI)
AuthFallback(rec, req, "DoesNotExist", &cfg.ClientAPI)
if rec.Code != http.StatusNotImplemented {
t.Fatalf("unexpected http status: %d, want %d", rec.Code, http.StatusNotImplemented)
}
@ -114,7 +112,7 @@ func Test_AuthFallback(t *testing.T) {
t.Run("unknown methods are handled correctly", func(t *testing.T) {
req := httptest.NewRequest(http.MethodDelete, "/?session=1337", nil)
rec := httptest.NewRecorder()
AuthFallback(rec, req, authtypes.LoginTypeRecaptcha, &base.Cfg.ClientAPI)
AuthFallback(rec, req, authtypes.LoginTypeRecaptcha, &cfg.ClientAPI)
if rec.Code != http.StatusMethodNotAllowed {
t.Fatalf("unexpected http status: %d, want %d", rec.Code, http.StatusMethodNotAllowed)
}
@ -123,7 +121,7 @@ func Test_AuthFallback(t *testing.T) {
t.Run("missing session parameter is handled correctly", func(t *testing.T) {
req := httptest.NewRequest(http.MethodGet, "/", nil)
rec := httptest.NewRecorder()
AuthFallback(rec, req, authtypes.LoginTypeRecaptcha, &base.Cfg.ClientAPI)
AuthFallback(rec, req, authtypes.LoginTypeRecaptcha, &cfg.ClientAPI)
if rec.Code != http.StatusBadRequest {
t.Fatalf("unexpected http status: %d, want %d", rec.Code, http.StatusBadRequest)
}
@ -132,7 +130,7 @@ func Test_AuthFallback(t *testing.T) {
t.Run("missing session parameter is handled correctly", func(t *testing.T) {
req := httptest.NewRequest(http.MethodGet, "/", nil)
rec := httptest.NewRecorder()
AuthFallback(rec, req, authtypes.LoginTypeRecaptcha, &base.Cfg.ClientAPI)
AuthFallback(rec, req, authtypes.LoginTypeRecaptcha, &cfg.ClientAPI)
if rec.Code != http.StatusBadRequest {
t.Fatalf("unexpected http status: %d, want %d", rec.Code, http.StatusBadRequest)
}
@ -141,7 +139,7 @@ func Test_AuthFallback(t *testing.T) {
t.Run("missing 'response' is handled correctly", func(t *testing.T) {
req := httptest.NewRequest(http.MethodPost, "/?session=1337", nil)
rec := httptest.NewRecorder()
AuthFallback(rec, req, authtypes.LoginTypeRecaptcha, &base.Cfg.ClientAPI)
AuthFallback(rec, req, authtypes.LoginTypeRecaptcha, &cfg.ClientAPI)
if rec.Code != http.StatusBadRequest {
t.Fatalf("unexpected http status: %d, want %d", rec.Code, http.StatusBadRequest)
}

8
cmd/dendrite-demo-pinecone/conn/client.go
View file

@ -21,7 +21,7 @@ import (
"net/http"
"strings"
"github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/gomatrixserverlib"
"nhooyr.io/websocket"
@ -90,7 +90,7 @@ func createTransport(s *pineconeSessions.Sessions) *http.Transport {
}
func CreateClient(
base *base.BaseDendrite, s *pineconeSessions.Sessions,
s *pineconeSessions.Sessions,
) *gomatrixserverlib.Client {
return gomatrixserverlib.NewClient(
gomatrixserverlib.WithTransport(createTransport(s)),
@ -98,10 +98,10 @@ func CreateClient(
}
func CreateFederationClient(
base *base.BaseDendrite, s *pineconeSessions.Sessions,
cfg *config.Dendrite, s *pineconeSessions.Sessions,
) *gomatrixserverlib.FederationClient {
return gomatrixserverlib.NewFederationClient(
base.Cfg.Global.SigningIdentities(),
cfg.Global.SigningIdentities(),
gomatrixserverlib.WithTransport(createTransport(s)),
)
}

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

@ -27,8 +27,11 @@ import (
"github.com/matrix-org/dendrite/cmd/dendrite-demo-pinecone/monolith"
"github.com/matrix-org/dendrite/cmd/dendrite-demo-yggdrasil/signing"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/setup"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/process"
"github.com/matrix-org/gomatrixserverlib"
"github.com/sirupsen/logrus"
@ -87,9 +90,13 @@ func main() {
}
}
processCtx := process.NewProcessContext()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
routers := httputil.NewRouters()
enableMetrics := true
enableWebsockets := true
p2pMonolith.SetupDendrite(cfg, *instancePort, *instanceRelayingEnabled, enableMetrics, enableWebsockets)
p2pMonolith.SetupDendrite(processCtx, cfg, cm, routers, *instancePort, *instanceRelayingEnabled, enableMetrics, enableWebsockets)
p2pMonolith.StartMonolith()
p2pMonolith.WaitForShutdown()

75
cmd/dendrite-demo-pinecone/monolith/monolith.go
View file

@ -39,6 +39,7 @@ import (
"github.com/matrix-org/dendrite/federationapi/producers"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/relayapi"
relayAPI "github.com/matrix-org/dendrite/relayapi/api"
"github.com/matrix-org/dendrite/roomserver"
@ -46,6 +47,7 @@ import (
"github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/setup/process"
"github.com/matrix-org/dendrite/userapi"
userAPI "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/gomatrixserverlib"
@ -61,13 +63,13 @@ import (
const SessionProtocol = "matrix"
type P2PMonolith struct {
BaseDendrite *base.BaseDendrite
Sessions *pineconeSessions.Sessions
Multicast *pineconeMulticast.Multicast
ConnManager *pineconeConnections.ConnectionManager
Router *pineconeRouter.Router
EventChannel chan pineconeEvents.Event
RelayRetriever relay.RelayServerRetriever
ProcessCtx *process.ProcessContext
dendrite setup.Monolith
port int
@ -121,54 +123,52 @@ func (p *P2PMonolith) SetupPinecone(sk ed25519.PrivateKey) {
p.ConnManager = pineconeConnections.NewConnectionManager(p.Router, nil)
}
func (p *P2PMonolith) SetupDendrite(cfg *config.Dendrite, port int, enableRelaying bool, enableMetrics bool, enableWebsockets bool) {
if enableMetrics {
p.BaseDendrite = base.NewBaseDendrite(cfg)
} else {
p.BaseDendrite = base.NewBaseDendrite(cfg, base.DisableMetrics)
}
p.port = port
p.BaseDendrite.ConfigureAdminEndpoints()
func (p *P2PMonolith) SetupDendrite(
processCtx *process.ProcessContext, cfg *config.Dendrite, cm sqlutil.Connections, routers httputil.Routers,
port int, enableRelaying bool, enableMetrics bool, enableWebsockets bool) {
federation := conn.CreateFederationClient(p.BaseDendrite, p.Sessions)
p.port = port
base.ConfigureAdminEndpoints(processCtx, routers)
federation := conn.CreateFederationClient(cfg, p.Sessions)
serverKeyAPI := &signing.YggdrasilKeys{}
keyRing := serverKeyAPI.KeyRing()
caches := caching.NewRistrettoCache(cfg.Global.Cache.EstimatedMaxSize, cfg.Global.Cache.MaxAge, enableMetrics)
natsInstance := jetstream.NATSInstance{}
rsAPI := roomserver.NewInternalAPI(p.BaseDendrite.ProcessContext, p.BaseDendrite.Cfg, p.BaseDendrite.ConnectionManager, &natsInstance, caches, p.BaseDendrite.EnableMetrics)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, enableMetrics)
fsAPI := federationapi.NewInternalAPI(
p.BaseDendrite.ProcessContext, p.BaseDendrite.Cfg, p.BaseDendrite.ConnectionManager, &natsInstance, federation, rsAPI, caches, keyRing, true,
processCtx, cfg, cm, &natsInstance, federation, rsAPI, caches, keyRing, true,
)
userAPI := userapi.NewInternalAPI(p.BaseDendrite.ProcessContext, p.BaseDendrite.Cfg, p.BaseDendrite.ConnectionManager, &natsInstance, rsAPI, federation)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, federation)
asAPI := appservice.NewInternalAPI(p.BaseDendrite.ProcessContext, p.BaseDendrite.Cfg, &natsInstance, userAPI, rsAPI)
asAPI := appservice.NewInternalAPI(processCtx, cfg, &natsInstance, userAPI, rsAPI)
rsAPI.SetFederationAPI(fsAPI, keyRing)
userProvider := users.NewPineconeUserProvider(p.Router, p.Sessions, userAPI, federation)
roomProvider := rooms.NewPineconeRoomProvider(p.Router, p.Sessions, fsAPI, federation)
js, _ := natsInstance.Prepare(p.BaseDendrite.ProcessContext, &p.BaseDendrite.Cfg.Global.JetStream)
js, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
producer := &producers.SyncAPIProducer{
JetStream: js,
TopicReceiptEvent: p.BaseDendrite.Cfg.Global.JetStream.Prefixed(jetstream.OutputReceiptEvent),
TopicSendToDeviceEvent: p.BaseDendrite.Cfg.Global.JetStream.Prefixed(jetstream.OutputSendToDeviceEvent),
TopicTypingEvent: p.BaseDendrite.Cfg.Global.JetStream.Prefixed(jetstream.OutputTypingEvent),
TopicPresenceEvent: p.BaseDendrite.Cfg.Global.JetStream.Prefixed(jetstream.OutputPresenceEvent),
TopicDeviceListUpdate: p.BaseDendrite.Cfg.Global.JetStream.Prefixed(jetstream.InputDeviceListUpdate),
TopicSigningKeyUpdate: p.BaseDendrite.Cfg.Global.JetStream.Prefixed(jetstream.InputSigningKeyUpdate),
Config: &p.BaseDendrite.Cfg.FederationAPI,
TopicReceiptEvent: cfg.Global.JetStream.Prefixed(jetstream.OutputReceiptEvent),
TopicSendToDeviceEvent: cfg.Global.JetStream.Prefixed(jetstream.OutputSendToDeviceEvent),
TopicTypingEvent: cfg.Global.JetStream.Prefixed(jetstream.OutputTypingEvent),
TopicPresenceEvent: cfg.Global.JetStream.Prefixed(jetstream.OutputPresenceEvent),
TopicDeviceListUpdate: cfg.Global.JetStream.Prefixed(jetstream.InputDeviceListUpdate),
TopicSigningKeyUpdate: cfg.Global.JetStream.Prefixed(jetstream.InputSigningKeyUpdate),
Config: &cfg.FederationAPI,
UserAPI: userAPI,
}
relayAPI := relayapi.NewRelayInternalAPI(p.BaseDendrite.Cfg, p.BaseDendrite.ConnectionManager, federation, rsAPI, keyRing, producer, enableRelaying, caches)
relayAPI := relayapi.NewRelayInternalAPI(cfg, cm, federation, rsAPI, keyRing, producer, enableRelaying, caches)
logrus.Infof("Relaying enabled: %v", relayAPI.RelayingEnabled())
p.dendrite = setup.Monolith{
Config: p.BaseDendrite.Cfg,
Client: conn.CreateClient(p.BaseDendrite, p.Sessions),
Config: cfg,
Client: conn.CreateClient(p.Sessions),
FedClient: federation,
KeyRing: keyRing,
@ -180,9 +180,10 @@ func (p *P2PMonolith) SetupDendrite(cfg *config.Dendrite, port int, enableRelayi
ExtPublicRoomsProvider: roomProvider,
ExtUserDirectoryProvider: userProvider,
}
p.dendrite.AddAllPublicRoutes(p.BaseDendrite, &natsInstance, caches)
p.ProcessCtx = processCtx
p.dendrite.AddAllPublicRoutes(processCtx, cfg, routers, cm, &natsInstance, caches, enableMetrics)
p.setupHttpServers(userProvider, enableWebsockets)
p.setupHttpServers(userProvider, routers, enableWebsockets)
}
func (p *P2PMonolith) GetFederationAPI() federationAPI.FederationInternalAPI {
@ -204,13 +205,13 @@ func (p *P2PMonolith) StartMonolith() {
func (p *P2PMonolith) Stop() {
logrus.Info("Stopping monolith")
_ = p.BaseDendrite.Close()
p.ProcessCtx.ShutdownDendrite()
p.WaitForShutdown()
logrus.Info("Stopped monolith")
}
func (p *P2PMonolith) WaitForShutdown() {
p.BaseDendrite.WaitForShutdown()
p.ProcessCtx.WaitForShutdown()
p.closeAllResources()
}
@ -247,12 +248,12 @@ func (p *P2PMonolith) Addr() string {
return p.httpListenAddr
}
func (p *P2PMonolith) setupHttpServers(userProvider *users.PineconeUserProvider, enableWebsockets bool) {
func (p *P2PMonolith) setupHttpServers(userProvider *users.PineconeUserProvider, routers httputil.Routers, enableWebsockets bool) {
p.httpMux = mux.NewRouter().SkipClean(true).UseEncodedPath()
p.httpMux.PathPrefix(httputil.PublicClientPathPrefix).Handler(p.BaseDendrite.Routers.Client)
p.httpMux.PathPrefix(httputil.PublicMediaPathPrefix).Handler(p.BaseDendrite.Routers.Media)
p.httpMux.PathPrefix(httputil.DendriteAdminPathPrefix).Handler(p.BaseDendrite.Routers.DendriteAdmin)
p.httpMux.PathPrefix(httputil.SynapseAdminPathPrefix).Handler(p.BaseDendrite.Routers.SynapseAdmin)
p.httpMux.PathPrefix(httputil.PublicClientPathPrefix).Handler(routers.Client)
p.httpMux.PathPrefix(httputil.PublicMediaPathPrefix).Handler(routers.Media)
p.httpMux.PathPrefix(httputil.DendriteAdminPathPrefix).Handler(routers.DendriteAdmin)
p.httpMux.PathPrefix(httputil.SynapseAdminPathPrefix).Handler(routers.SynapseAdmin)
if enableWebsockets {
wsUpgrader := websocket.Upgrader{
@ -285,8 +286,8 @@ func (p *P2PMonolith) setupHttpServers(userProvider *users.PineconeUserProvider,
p.pineconeMux = mux.NewRouter().SkipClean(true).UseEncodedPath()
p.pineconeMux.PathPrefix(users.PublicURL).HandlerFunc(userProvider.FederatedUserProfiles)
p.pineconeMux.PathPrefix(httputil.PublicFederationPathPrefix).Handler(p.BaseDendrite.Routers.Federation)
p.pineconeMux.PathPrefix(httputil.PublicMediaPathPrefix).Handler(p.BaseDendrite.Routers.Media)
p.pineconeMux.PathPrefix(httputil.PublicFederationPathPrefix).Handler(routers.Federation)
p.pineconeMux.PathPrefix(httputil.PublicMediaPathPrefix).Handler(routers.Media)
pHTTP := p.Sessions.Protocol(SessionProtocol).HTTP()
pHTTP.Mux().Handle(users.PublicURL, p.pineconeMux)
@ -370,7 +371,7 @@ func (p *P2PMonolith) startEventHandler() {
ServerNames: []gomatrixserverlib.ServerName{gomatrixserverlib.ServerName(e.PeerID)},
}
res := &federationAPI.PerformWakeupServersResponse{}
if err := p.dendrite.FederationAPI.PerformWakeupServers(p.BaseDendrite.Context(), req, res); err != nil {
if err := p.dendrite.FederationAPI.PerformWakeupServers(p.ProcessCtx.Context(), req, res); err != nil {
eLog.WithError(err).Error("Failed to wakeup destination", e.PeerID)
}
}

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

@ -27,8 +27,11 @@ import (
"path/filepath"
"time"
"github.com/getsentry/sentry-go"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/setup/process"
"github.com/matrix-org/gomatrixserverlib"
"github.com/gorilla/mux"
@ -43,7 +46,7 @@ import (
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/roomserver"
"github.com/matrix-org/dendrite/setup"
"github.com/matrix-org/dendrite/setup/base"
basepkg "github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/mscs"
"github.com/matrix-org/dendrite/test"
@ -144,37 +147,83 @@ func main() {
cfg.Global.ServerName = gomatrixserverlib.ServerName(hex.EncodeToString(pk))
cfg.Global.KeyID = gomatrixserverlib.KeyID(signing.KeyID)
base := base.NewBaseDendrite(cfg)
base.ConfigureAdminEndpoints()
defer base.Close() // nolint: errcheck
configErrors := &config.ConfigErrors{}
cfg.Verify(configErrors)
if len(*configErrors) > 0 {
for _, err := range *configErrors {
logrus.Errorf("Configuration error: %s", err)
}
logrus.Fatalf("Failed to start due to configuration errors")
}
internal.SetupStdLogging()
internal.SetupHookLogging(cfg.Logging)
internal.SetupPprof()
logrus.Infof("Dendrite version %s", internal.VersionString())
if !cfg.ClientAPI.RegistrationDisabled && cfg.ClientAPI.OpenRegistrationWithoutVerificationEnabled {
logrus.Warn("Open registration is enabled")
}
closer, err := cfg.SetupTracing()
if err != nil {
logrus.WithError(err).Panicf("failed to start opentracing")
}
defer closer.Close()
if cfg.Global.Sentry.Enabled {
logrus.Info("Setting up Sentry for debugging...")
err = sentry.Init(sentry.ClientOptions{
Dsn: cfg.Global.Sentry.DSN,
Environment: cfg.Global.Sentry.Environment,
Debug: true,
ServerName: string(cfg.Global.ServerName),
Release: "dendrite@" + internal.VersionString(),
AttachStacktrace: true,
})
if err != nil {
logrus.WithError(err).Panic("failed to start Sentry")
}
}
processCtx := process.NewProcessContext()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
routers := httputil.NewRouters()
basepkg.ConfigureAdminEndpoints(processCtx, routers)
defer func() {
processCtx.ShutdownDendrite()
processCtx.WaitForShutdown()
}() // nolint: errcheck
ygg, err := yggconn.Setup(sk, *instanceName, ".", *instancePeer, *instanceListen)
if err != nil {
panic(err)
}
federation := ygg.CreateFederationClient(base)
federation := ygg.CreateFederationClient(cfg)
serverKeyAPI := &signing.YggdrasilKeys{}
keyRing := serverKeyAPI.KeyRing()
caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.EnableMetrics)
caches := caching.NewRistrettoCache(cfg.Global.Cache.EstimatedMaxSize, cfg.Global.Cache.MaxAge, caching.EnableMetrics)
natsInstance := jetstream.NATSInstance{}
rsAPI := roomserver.NewInternalAPI(base.ProcessContext, base.Cfg, base.ConnectionManager, &natsInstance, caches, base.EnableMetrics)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.EnableMetrics)
userAPI := userapi.NewInternalAPI(base.ProcessContext, base.Cfg, base.ConnectionManager, &natsInstance, rsAPI, federation)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, federation)
asAPI := appservice.NewInternalAPI(base.ProcessContext, base.Cfg, &natsInstance, userAPI, rsAPI)
asAPI := appservice.NewInternalAPI(processCtx, cfg, &natsInstance, userAPI, rsAPI)
rsAPI.SetAppserviceAPI(asAPI)
fsAPI := federationapi.NewInternalAPI(
base.ProcessContext, base.Cfg, base.ConnectionManager, &natsInstance, federation, rsAPI, caches, keyRing, true,
processCtx, cfg, cm, &natsInstance, federation, rsAPI, caches, keyRing, true,
)
rsAPI.SetFederationAPI(fsAPI, keyRing)
monolith := setup.Monolith{
Config: base.Cfg,
Client: ygg.CreateClient(base),
Config: cfg,
Client: ygg.CreateClient(),
FedClient: federation,
KeyRing: keyRing,
@ -186,21 +235,21 @@ func main() {
ygg, fsAPI, federation,
),
}
monolith.AddAllPublicRoutes(base, &natsInstance, caches)
if err := mscs.Enable(base, &monolith, caches); err != nil {
monolith.AddAllPublicRoutes(processCtx, cfg, routers, cm, &natsInstance, caches, caching.EnableMetrics)
if err := mscs.Enable(cfg, cm, routers, &monolith, caches); err != nil {
logrus.WithError(err).Fatalf("Failed to enable MSCs")
}
httpRouter := mux.NewRouter().SkipClean(true).UseEncodedPath()
httpRouter.PathPrefix(httputil.PublicClientPathPrefix).Handler(base.Routers.Client)
httpRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(base.Routers.Media)
httpRouter.PathPrefix(httputil.DendriteAdminPathPrefix).Handler(base.Routers.DendriteAdmin)
httpRouter.PathPrefix(httputil.SynapseAdminPathPrefix).Handler(base.Routers.SynapseAdmin)
httpRouter.PathPrefix(httputil.PublicClientPathPrefix).Handler(routers.Client)
httpRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(routers.Media)
httpRouter.PathPrefix(httputil.DendriteAdminPathPrefix).Handler(routers.DendriteAdmin)
httpRouter.PathPrefix(httputil.SynapseAdminPathPrefix).Handler(routers.SynapseAdmin)
embed.Embed(httpRouter, *instancePort, "Yggdrasil Demo")
yggRouter := mux.NewRouter().SkipClean(true).UseEncodedPath()
yggRouter.PathPrefix(httputil.PublicFederationPathPrefix).Handler(base.Routers.Federation)
yggRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(base.Routers.Media)
yggRouter.PathPrefix(httputil.PublicFederationPathPrefix).Handler(routers.Federation)
yggRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(routers.Media)
// Build both ends of a HTTP multiplex.
httpServer := &http.Server{
@ -234,5 +283,5 @@ func main() {
}()
// We want to block forever to let the HTTP and HTTPS handler serve the APIs
base.WaitForShutdown()
basepkg.WaitForShutdown(processCtx)
}

10
cmd/dendrite-demo-yggdrasil/yggconn/client.go
View file

@ -4,7 +4,7 @@ import (
"net/http"
"time"
"github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/gomatrixserverlib"
)
@ -17,9 +17,7 @@ func (y *yggroundtripper) RoundTrip(req *http.Request) (*http.Response, error) {
return y.inner.RoundTrip(req)
}
func (n *Node) CreateClient(
base *base.BaseDendrite,
) *gomatrixserverlib.Client {
func (n *Node) CreateClient() *gomatrixserverlib.Client {
tr := &http.Transport{}
tr.RegisterProtocol(
"matrix", &yggroundtripper{
@ -39,7 +37,7 @@ func (n *Node) CreateClient(
}
func (n *Node) CreateFederationClient(
base *base.BaseDendrite,
cfg *config.Dendrite,
) *gomatrixserverlib.FederationClient {
tr := &http.Transport{}
tr.RegisterProtocol(
@ -55,7 +53,7 @@ func (n *Node) CreateFederationClient(
},
)
return gomatrixserverlib.NewFederationClient(
base.Cfg.Global.SigningIdentities(),
cfg.Global.SigningIdentities(),
gomatrixserverlib.WithTransport(tr),
)
}

110
cmd/dendrite/main.go
View file

@ -16,9 +16,16 @@ package main
import (
"flag"
"time"
"github.com/getsentry/sentry-go"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/setup/process"
"github.com/matrix-org/gomatrixserverlib"
"github.com/sirupsen/logrus"
"github.com/matrix-org/dendrite/appservice"
@ -67,25 +74,90 @@ func main() {
httpAddr = socket
}
options := []basepkg.BaseDendriteOptions{}
configErrors := &config.ConfigErrors{}
cfg.Verify(configErrors)
if len(*configErrors) > 0 {
for _, err := range *configErrors {
logrus.Errorf("Configuration error: %s", err)
}
logrus.Fatalf("Failed to start due to configuration errors")
}
processCtx := process.NewProcessContext()
base := basepkg.NewBaseDendrite(cfg, options...)
defer base.Close() // nolint: errcheck
internal.SetupStdLogging()
internal.SetupHookLogging(cfg.Logging)
internal.SetupPprof()
federation := base.CreateFederationClient()
basepkg.PlatformSanityChecks()
caches := caching.NewRistrettoCache(base.Cfg.Global.Cache.EstimatedMaxSize, base.Cfg.Global.Cache.MaxAge, caching.EnableMetrics)
logrus.Infof("Dendrite version %s", internal.VersionString())
if !cfg.ClientAPI.RegistrationDisabled && cfg.ClientAPI.OpenRegistrationWithoutVerificationEnabled {
logrus.Warn("Open registration is enabled")
}
// create DNS cache
var dnsCache *gomatrixserverlib.DNSCache
if cfg.Global.DNSCache.Enabled {
dnsCache = gomatrixserverlib.NewDNSCache(
cfg.Global.DNSCache.CacheSize,
cfg.Global.DNSCache.CacheLifetime,
)
logrus.Infof(
"DNS cache enabled (size %d, lifetime %s)",
cfg.Global.DNSCache.CacheSize,
cfg.Global.DNSCache.CacheLifetime,
)
}
// setup tracing
closer, err := cfg.SetupTracing()
if err != nil {
logrus.WithError(err).Panicf("failed to start opentracing")
}
defer closer.Close()
// setup sentry
if cfg.Global.Sentry.Enabled {
logrus.Info("Setting up Sentry for debugging...")
err = sentry.Init(sentry.ClientOptions{
Dsn: cfg.Global.Sentry.DSN,
Environment: cfg.Global.Sentry.Environment,
Debug: true,
ServerName: string(cfg.Global.ServerName),
Release: "dendrite@" + internal.VersionString(),
AttachStacktrace: true,
})
if err != nil {
logrus.WithError(err).Panic("failed to start Sentry")
}
go func() {
processCtx.ComponentStarted()
<-processCtx.WaitForShutdown()
if !sentry.Flush(time.Second * 5) {
logrus.Warnf("failed to flush all Sentry events!")
}
processCtx.ComponentFinished()
}()
}
federationClient := basepkg.CreateFederationClient(cfg, dnsCache)
httpClient := basepkg.CreateClient(cfg, dnsCache)
// prepare required dependencies
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
routers := httputil.NewRouters()
caches := caching.NewRistrettoCache(cfg.Global.Cache.EstimatedMaxSize, cfg.Global.Cache.MaxAge, caching.EnableMetrics)
natsInstance := jetstream.NATSInstance{}
rsAPI := roomserver.NewInternalAPI(base.ProcessContext, base.Cfg, base.ConnectionManager, &natsInstance, caches, base.EnableMetrics)
rsAPI := roomserver.NewInternalAPI(processCtx, cfg, cm, &natsInstance, caches, caching.EnableMetrics)
fsAPI := federationapi.NewInternalAPI(
base.ProcessContext, base.Cfg, base.ConnectionManager, &natsInstance, federation, rsAPI, caches, nil, false,
processCtx, cfg, cm, &natsInstance, federationClient, rsAPI, caches, nil, false,
)
keyRing := fsAPI.KeyRing()
userAPI := userapi.NewInternalAPI(base.ProcessContext, base.Cfg, base.ConnectionManager, &natsInstance, rsAPI, federation)
asAPI := appservice.NewInternalAPI(base.ProcessContext, base.Cfg, &natsInstance, userAPI, rsAPI)
userAPI := userapi.NewInternalAPI(processCtx, cfg, cm, &natsInstance, rsAPI, federationClient)
asAPI := appservice.NewInternalAPI(processCtx, cfg, &natsInstance, userAPI, rsAPI)
// The underlying roomserver implementation needs to be able to call the fedsender.
// This is different to rsAPI which can be the http client which doesn't need this
@ -95,9 +167,9 @@ func main() {
rsAPI.SetUserAPI(userAPI)
monolith := setup.Monolith{
Config: base.Cfg,
Client: base.CreateClient(),
FedClient: federation,
Config: cfg,
Client: httpClient,
FedClient: federationClient,
KeyRing: keyRing,
AppserviceAPI: asAPI,
@ -107,25 +179,25 @@ func main() {
RoomserverAPI: rsAPI,
UserAPI: userAPI,
}
monolith.AddAllPublicRoutes(base, &natsInstance, caches)
monolith.AddAllPublicRoutes(processCtx, cfg, routers, cm, &natsInstance, caches, caching.EnableMetrics)
if len(base.Cfg.MSCs.MSCs) > 0 {
if err := mscs.Enable(base, &monolith, caches); err != nil {
if len(cfg.MSCs.MSCs) > 0 {
if err := mscs.Enable(cfg, cm, routers, &monolith, caches); err != nil {
logrus.WithError(err).Fatalf("Failed to enable MSCs")
}
}
// Expose the matrix APIs directly rather than putting them under a /api path.
go func() {
base.SetupAndServeHTTP(httpAddr, nil, nil)
basepkg.SetupAndServeHTTP(processCtx, cfg, routers, httpAddr, nil, nil)
}()
// Handle HTTPS if certificate and key are provided
if *unixSocket == "" && *certFile != "" && *keyFile != "" {
go func() {
base.SetupAndServeHTTP(httpsAddr, certFile, keyFile)
basepkg.SetupAndServeHTTP(processCtx, cfg, routers, httpsAddr, certFile, keyFile)
}()
}
// We want to block forever to let the HTTP and HTTPS handler serve the APIs
base.WaitForShutdown()
basepkg.WaitForShutdown(processCtx)
}

9
cmd/resolve-state/main.go
View file

@ -10,12 +10,13 @@ import (
"time"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver/state"
"github.com/matrix-org/dendrite/roomserver/storage"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/dendrite/setup"
"github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/process"
"github.com/matrix-org/gomatrixserverlib"
)
@ -40,7 +41,7 @@ func main() {
Level: "error",
})
cfg.ClientAPI.RegistrationDisabled = true
base := base.NewBaseDendrite(cfg, base.DisableMetrics)
args := flag.Args()
fmt.Println("Room version", *roomVersion)
@ -54,8 +55,10 @@ func main() {
fmt.Println("Fetching", len(snapshotNIDs), "snapshot NIDs")
processCtx := process.NewProcessContext()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
roomserverDB, err := storage.Open(
base.ProcessContext.Context(), base.ConnectionManager, &cfg.RoomServer.Database,
processCtx.Context(), cm, &cfg.RoomServer.Database,
caching.NewRistrettoCache(128*1024*1024, time.Hour, true),
)
if err != nil {

8
federationapi/federationapi_keys_test.go
View file

@ -12,8 +12,9 @@ import (
"testing"
"time"
"github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/setup/process"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/dendrite/federationapi/api"
@ -110,8 +111,9 @@ func TestMain(m *testing.M) {
)
// Finally, build the server key APIs.
sbase := base.NewBaseDendrite(cfg, base.DisableMetrics)
s.api = NewInternalAPI(sbase.ProcessContext, sbase.Cfg, sbase.ConnectionManager, &natsInstance, s.fedclient, nil, s.cache, nil, true)
processCtx := process.NewProcessContext()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
s.api = NewInternalAPI(processCtx, cfg, cm, &natsInstance, s.fedclient, nil, s.cache, nil, true)
}
// Now that we have built our server key APIs, start the

8
roomserver/roomserver_test.go
View file

@ -230,7 +230,9 @@ func TestPurgeRoom(t *testing.T) {
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
caches := caching.NewRistrettoCache(128*1024*1024, time.Hour, caching.DisableMetrics)
db, err := storage.Open(processCtx.Context(), cm, &cfg.RoomServer.Database, caches)
if err != nil {
t.Fatal(err)
}
jsCtx, _ := natsInstance.Prepare(processCtx, &cfg.Global.JetStream)
defer jetstream.DeleteAllStreams(jsCtx, &cfg.Global.JetStream)
@ -243,13 +245,13 @@ func TestPurgeRoom(t *testing.T) {
rsAPI.SetFederationAPI(nil, nil)
// Create the room
if err := api.SendEvents(ctx, rsAPI, api.KindNew, room.Events(), "test", "test", "test", nil, false); err != nil {
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)
}
// some dummy entries to validate after purging
publishResp := &api.PerformPublishResponse{}
if err := rsAPI.PerformPublish(ctx, &api.PerformPublishRequest{RoomID: room.ID, Visibility: "public"}, publishResp); err != nil {
if err = rsAPI.PerformPublish(ctx, &api.PerformPublishRequest{RoomID: room.ID, Visibility: "public"}, publishResp); err != nil {
t.Fatal(err)
}
if publishResp.Error != nil {

4
roomserver/storage/shared/storage_test.go
View file

@ -15,14 +15,12 @@ import (
"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())
@ -61,8 +59,6 @@ func mustCreateRoomserverDatabase(t *testing.T, dbType test.DBType) (*shared.Dat
Writer: sqlutil.NewExclusiveWriter(),
Cache: cache,
}, func() {
err := base.Close()
assert.NoError(t, err)
clearDB()
err = db.Close()
assert.NoError(t, err)

230
setup/base/base.go
View file

@ -22,29 +22,24 @@ import (
"errors"
"fmt"
"html/template"
"io"
"io/fs"
"net"
"net/http"
_ "net/http/pprof"
"os"
"os/signal"
"sync"
"syscall"
"time"
"github.com/getsentry/sentry-go"
sentryhttp "github.com/getsentry/sentry-go/http"
"github.com/matrix-org/gomatrixserverlib"
"github.com/prometheus/client_golang/prometheus/promhttp"
"go.uber.org/atomic"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/gorilla/mux"
"github.com/kardianos/minwinsvc"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/sirupsen/logrus"
@ -55,143 +50,22 @@ import (
//go:embed static/*.gotmpl
var staticContent embed.FS
// BaseDendrite is a base for creating new instances of dendrite. It parses
// command line flags and config, and exposes methods for creating various
// resources. All errors are handled by logging then exiting, so all methods
// should only be used during start up.
// Must be closed when shutting down.
type BaseDendrite struct {
*process.ProcessContext
tracerCloser io.Closer
Routers httputil.Routers
Cfg *config.Dendrite
DNSCache *gomatrixserverlib.DNSCache
ConnectionManager sqlutil.Connections
EnableMetrics bool
startupLock sync.Mutex
}
const HTTPServerTimeout = time.Minute * 5
type BaseDendriteOptions int
const (
DisableMetrics BaseDendriteOptions = iota
)
// NewBaseDendrite creates a new instance to be used by a component.
func NewBaseDendrite(cfg *config.Dendrite, options ...BaseDendriteOptions) *BaseDendrite {
platformSanityChecks()
enableMetrics := true
for _, opt := range options {
switch opt {
case DisableMetrics:
enableMetrics = false
}
}
configErrors := &config.ConfigErrors{}
cfg.Verify(configErrors)
if len(*configErrors) > 0 {
for _, err := range *configErrors {
logrus.Errorf("Configuration error: %s", err)
}
logrus.Fatalf("Failed to start due to configuration errors")
}
internal.SetupStdLogging()
internal.SetupHookLogging(cfg.Logging)
internal.SetupPprof()
logrus.Infof("Dendrite version %s", internal.VersionString())
if !cfg.ClientAPI.RegistrationDisabled && cfg.ClientAPI.OpenRegistrationWithoutVerificationEnabled {
logrus.Warn("Open registration is enabled")
}
closer, err := cfg.SetupTracing()
if err != nil {
logrus.WithError(err).Panicf("failed to start opentracing")
}
if cfg.Global.Sentry.Enabled {
logrus.Info("Setting up Sentry for debugging...")
err = sentry.Init(sentry.ClientOptions{
Dsn: cfg.Global.Sentry.DSN,
Environment: cfg.Global.Sentry.Environment,
Debug: true,
ServerName: string(cfg.Global.ServerName),
Release: "dendrite@" + internal.VersionString(),
AttachStacktrace: true,
})
if err != nil {
logrus.WithError(err).Panic("failed to start Sentry")
}
}
var dnsCache *gomatrixserverlib.DNSCache
if cfg.Global.DNSCache.Enabled {
dnsCache = gomatrixserverlib.NewDNSCache(
cfg.Global.DNSCache.CacheSize,
cfg.Global.DNSCache.CacheLifetime,
)
logrus.Infof(
"DNS cache enabled (size %d, lifetime %s)",
cfg.Global.DNSCache.CacheSize,
cfg.Global.DNSCache.CacheLifetime,
)
}
// If we're in monolith mode, we'll set up a global pool of database
// connections. A component is welcome to use this pool if they don't
// have a separate database config of their own.
pCtx := process.NewProcessContext()
cm := sqlutil.NewConnectionManager(pCtx, cfg.Global.DatabaseOptions)
// Ideally we would only use SkipClean on routes which we know can allow '/' but due to
// https://github.com/gorilla/mux/issues/460 we have to attach this at the top router.
// When used in conjunction with UseEncodedPath() we get the behaviour we want when parsing
// path parameters:
// /foo/bar%2Fbaz == [foo, bar%2Fbaz] (from UseEncodedPath)
// /foo/bar%2F%2Fbaz == [foo, bar%2F%2Fbaz] (from SkipClean)
// In particular, rooms v3 event IDs are not urlsafe and can include '/' and because they
// are randomly generated it results in flakey tests.
// We need to be careful with media APIs if they read from a filesystem to make sure they
// are not inadvertently reading paths without cleaning, else this could introduce a
// directory traversal attack e.g /../../../etc/passwd
return &BaseDendrite{
ProcessContext: pCtx,
tracerCloser: closer,
Cfg: cfg,
DNSCache: dnsCache,
Routers: httputil.NewRouters(),
ConnectionManager: cm,
EnableMetrics: enableMetrics,
}
}
// Close implements io.Closer
func (b *BaseDendrite) Close() error {
b.ProcessContext.ShutdownDendrite()
b.ProcessContext.WaitForShutdown()
return b.tracerCloser.Close()
}
// CreateClient creates a new client (normally used for media fetch requests).
// Should only be called once per component.
func (b *BaseDendrite) CreateClient() *gomatrixserverlib.Client {
if b.Cfg.Global.DisableFederation {
func CreateClient(cfg *config.Dendrite, dnsCache *gomatrixserverlib.DNSCache) *gomatrixserverlib.Client {
if cfg.Global.DisableFederation {
return gomatrixserverlib.NewClient(
gomatrixserverlib.WithTransport(noOpHTTPTransport),
)
}
opts := []gomatrixserverlib.ClientOption{
gomatrixserverlib.WithSkipVerify(b.Cfg.FederationAPI.DisableTLSValidation),
gomatrixserverlib.WithSkipVerify(cfg.FederationAPI.DisableTLSValidation),
gomatrixserverlib.WithWellKnownSRVLookups(true),
}
if b.Cfg.Global.DNSCache.Enabled {
opts = append(opts, gomatrixserverlib.WithDNSCache(b.DNSCache))
if cfg.Global.DNSCache.Enabled && dnsCache != nil {
opts = append(opts, gomatrixserverlib.WithDNSCache(dnsCache))
}
client := gomatrixserverlib.NewClient(opts...)
client.SetUserAgent(fmt.Sprintf("Dendrite/%s", internal.VersionString()))
@ -200,20 +74,20 @@ func (b *BaseDendrite) CreateClient() *gomatrixserverlib.Client {
// CreateFederationClient creates a new federation client. Should only be called
// once per component.
func (b *BaseDendrite) CreateFederationClient() *gomatrixserverlib.FederationClient {
identities := b.Cfg.Global.SigningIdentities()
if b.Cfg.Global.DisableFederation {
func CreateFederationClient(cfg *config.Dendrite, dnsCache *gomatrixserverlib.DNSCache) *gomatrixserverlib.FederationClient {
identities := cfg.Global.SigningIdentities()
if cfg.Global.DisableFederation {
return gomatrixserverlib.NewFederationClient(
identities, gomatrixserverlib.WithTransport(noOpHTTPTransport),
)
}
opts := []gomatrixserverlib.ClientOption{
gomatrixserverlib.WithTimeout(time.Minute * 5),
gomatrixserverlib.WithSkipVerify(b.Cfg.FederationAPI.DisableTLSValidation),
gomatrixserverlib.WithKeepAlives(!b.Cfg.FederationAPI.DisableHTTPKeepalives),
gomatrixserverlib.WithSkipVerify(cfg.FederationAPI.DisableTLSValidation),
gomatrixserverlib.WithKeepAlives(!cfg.FederationAPI.DisableHTTPKeepalives),
}
if b.Cfg.Global.DNSCache.Enabled {
opts = append(opts, gomatrixserverlib.WithDNSCache(b.DNSCache))
if cfg.Global.DNSCache.Enabled {
opts = append(opts, gomatrixserverlib.WithDNSCache(dnsCache))
}
client := gomatrixserverlib.NewFederationClient(
identities, opts...,
@ -222,12 +96,12 @@ func (b *BaseDendrite) CreateFederationClient() *gomatrixserverlib.FederationCli
return client
}
func (b *BaseDendrite) ConfigureAdminEndpoints() {
b.Routers.DendriteAdmin.HandleFunc("/monitor/up", func(w http.ResponseWriter, r *http.Request) {
func ConfigureAdminEndpoints(processContext *process.ProcessContext, routers httputil.Routers) {
routers.DendriteAdmin.HandleFunc("/monitor/up", func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(200)
})
b.Routers.DendriteAdmin.HandleFunc("/monitor/health", func(w http.ResponseWriter, r *http.Request) {
if isDegraded, reasons := b.ProcessContext.IsDegraded(); isDegraded {
routers.DendriteAdmin.HandleFunc("/monitor/health", func(w http.ResponseWriter, r *http.Request) {
if isDegraded, reasons := processContext.IsDegraded(); isDegraded {
w.WriteHeader(503)
_ = json.NewEncoder(w).Encode(struct {
Warnings []string `json:"warnings"`
@ -242,14 +116,13 @@ func (b *BaseDendrite) ConfigureAdminEndpoints() {
// SetupAndServeHTTP sets up the HTTP server to serve client & federation APIs
// and adds a prometheus handler under /_dendrite/metrics.
func (b *BaseDendrite) SetupAndServeHTTP(
func SetupAndServeHTTP(
processContext *process.ProcessContext,
cfg *config.Dendrite,
routers httputil.Routers,
externalHTTPAddr config.ServerAddress,
certFile, keyFile *string,
) {
// Manually unlocked right before actually serving requests,
// as we don't return from this method (defer doesn't work).
b.startupLock.Lock()
externalRouter := mux.NewRouter().SkipClean(true).UseEncodedPath()
externalServ := &http.Server{
@ -257,7 +130,7 @@ func (b *BaseDendrite) SetupAndServeHTTP(
WriteTimeout: HTTPServerTimeout,
Handler: externalRouter,
BaseContext: func(_ net.Listener) context.Context {
return b.ProcessContext.Context()
return processContext.Context()
},
}
@ -266,11 +139,11 @@ func (b *BaseDendrite) SetupAndServeHTTP(
http.Redirect(w, r, httputil.PublicStaticPath, http.StatusFound)
})
if b.Cfg.Global.Metrics.Enabled {
externalRouter.Handle("/metrics", httputil.WrapHandlerInBasicAuth(promhttp.Handler(), b.Cfg.Global.Metrics.BasicAuth))
if cfg.Global.Metrics.Enabled {
externalRouter.Handle("/metrics", httputil.WrapHandlerInBasicAuth(promhttp.Handler(), cfg.Global.Metrics.BasicAuth))
}
b.ConfigureAdminEndpoints()
ConfigureAdminEndpoints(processContext, routers)
// Parse and execute the landing page template
tmpl := template.Must(template.ParseFS(staticContent, "static/*.gotmpl"))
@ -281,38 +154,36 @@ func (b *BaseDendrite) SetupAndServeHTTP(
logrus.WithError(err).Fatal("failed to execute landing page template")
}
b.Routers.Static.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
routers.Static.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
_, _ = w.Write(landingPage.Bytes())
})
var clientHandler http.Handler
clientHandler = b.Routers.Client
if b.Cfg.Global.Sentry.Enabled {
clientHandler = routers.Client
if cfg.Global.Sentry.Enabled {
sentryHandler := sentryhttp.New(sentryhttp.Options{
Repanic: true,
})
clientHandler = sentryHandler.Handle(b.Routers.Client)
clientHandler = sentryHandler.Handle(routers.Client)
}
var federationHandler http.Handler
federationHandler = b.Routers.Federation
if b.Cfg.Global.Sentry.Enabled {
federationHandler = routers.Federation
if cfg.Global.Sentry.Enabled {
sentryHandler := sentryhttp.New(sentryhttp.Options{
Repanic: true,
})
federationHandler = sentryHandler.Handle(b.Routers.Federation)
federationHandler = sentryHandler.Handle(routers.Federation)
}
externalRouter.PathPrefix(httputil.DendriteAdminPathPrefix).Handler(b.Routers.DendriteAdmin)
externalRouter.PathPrefix(httputil.DendriteAdminPathPrefix).Handler(routers.DendriteAdmin)
externalRouter.PathPrefix(httputil.PublicClientPathPrefix).Handler(clientHandler)
if !b.Cfg.Global.DisableFederation {
externalRouter.PathPrefix(httputil.PublicKeyPathPrefix).Handler(b.Routers.Keys)
if !cfg.Global.DisableFederation {
externalRouter.PathPrefix(httputil.PublicKeyPathPrefix).Handler(routers.Keys)
externalRouter.PathPrefix(httputil.PublicFederationPathPrefix).Handler(federationHandler)
}
externalRouter.PathPrefix(httputil.SynapseAdminPathPrefix).Handler(b.Routers.SynapseAdmin)
externalRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(b.Routers.Media)
externalRouter.PathPrefix(httputil.PublicWellKnownPrefix).Handler(b.Routers.WellKnown)
externalRouter.PathPrefix(httputil.PublicStaticPath).Handler(b.Routers.Static)
b.startupLock.Unlock()
externalRouter.PathPrefix(httputil.SynapseAdminPathPrefix).Handler(routers.SynapseAdmin)
externalRouter.PathPrefix(httputil.PublicMediaPathPrefix).Handler(routers.Media)
externalRouter.PathPrefix(httputil.PublicWellKnownPrefix).Handler(routers.WellKnown)
externalRouter.PathPrefix(httputil.PublicStaticPath).Handler(routers.Static)
externalRouter.NotFoundHandler = httputil.NotFoundCORSHandler
externalRouter.MethodNotAllowedHandler = httputil.NotAllowedHandler
@ -321,10 +192,10 @@ func (b *BaseDendrite) SetupAndServeHTTP(
go func() {
var externalShutdown atomic.Bool // RegisterOnShutdown can be called more than once
logrus.Infof("Starting external listener on %s", externalServ.Addr)
b.ProcessContext.ComponentStarted()
processContext.ComponentStarted()
externalServ.RegisterOnShutdown(func() {
if externalShutdown.CompareAndSwap(false, true) {
b.ProcessContext.ComponentFinished()
processContext.ComponentFinished()
logrus.Infof("Stopped external HTTP listener")
}
})
@ -366,32 +237,27 @@ func (b *BaseDendrite) SetupAndServeHTTP(
}()
}
minwinsvc.SetOnExit(b.ProcessContext.ShutdownDendrite)
<-b.ProcessContext.WaitForShutdown()
minwinsvc.SetOnExit(processContext.ShutdownDendrite)
<-processContext.WaitForShutdown()
logrus.Infof("Stopping HTTP listeners")
_ = externalServ.Shutdown(context.Background())
logrus.Infof("Stopped HTTP listeners")
}
func (b *BaseDendrite) WaitForShutdown() {
func WaitForShutdown(processCtx *process.ProcessContext) {
sigs := make(chan os.Signal, 1)
signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
select {
case <-sigs:
case <-b.ProcessContext.WaitForShutdown():
case <-processCtx.WaitForShutdown():
}
signal.Reset(syscall.SIGINT, syscall.SIGTERM)
logrus.Warnf("Shutdown signal received")
b.ProcessContext.ShutdownDendrite()
b.ProcessContext.WaitForComponentsToFinish()
if b.Cfg.Global.Sentry.Enabled {
if !sentry.Flush(time.Second * 5) {
logrus.Warnf("failed to flush all Sentry events!")
}
}
processCtx.ShutdownDendrite()
processCtx.WaitForComponentsToFinish()
logrus.Warnf("Dendrite is exiting now")
}

20
setup/base/base_test.go
View file

@ -13,8 +13,10 @@ import (
"time"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/httputil"
basepkg "github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/test/testrig"
"github.com/matrix-org/dendrite/setup/process"
"github.com/stretchr/testify/assert"
)
@ -30,8 +32,10 @@ func TestLandingPage_Tcp(t *testing.T) {
})
assert.NoError(t, err)
b, _, _ := testrig.Base(nil)
defer b.Close()
processCtx := process.NewProcessContext()
routers := httputil.NewRouters()
cfg := config.Dendrite{}
cfg.Defaults(config.DefaultOpts{Generate: true, SingleDatabase: true})
// hack: create a server and close it immediately, just to get a random port assigned
s := httptest.NewServer(nil)
@ -40,7 +44,7 @@ func TestLandingPage_Tcp(t *testing.T) {
// start base with the listener and wait for it to be started
address, err := config.HTTPAddress(s.URL)
assert.NoError(t, err)
go b.SetupAndServeHTTP(address, nil, nil)
go basepkg.SetupAndServeHTTP(processCtx, &cfg, routers, address, nil, nil)
time.Sleep(time.Millisecond * 10)
// When hitting /, we should be redirected to /_matrix/static, which should contain the landing page
@ -70,15 +74,17 @@ func TestLandingPage_UnixSocket(t *testing.T) {
})
assert.NoError(t, err)
b, _, _ := testrig.Base(nil)
defer b.Close()
processCtx := process.NewProcessContext()
routers := httputil.NewRouters()
cfg := config.Dendrite{}
cfg.Defaults(config.DefaultOpts{Generate: true, SingleDatabase: true})
tempDir := t.TempDir()
socket := path.Join(tempDir, "socket")
// start base with the listener and wait for it to be started
address, err := config.UnixSocketAddress(socket, "755")
assert.NoError(t, err)
go b.SetupAndServeHTTP(address, nil, nil)
go basepkg.SetupAndServeHTTP(processCtx, &cfg, routers, address, nil, nil)
time.Sleep(time.Millisecond * 100)
client := &http.Client{

2
setup/base/sanity_other.go
View file

@ -3,6 +3,6 @@
package base
func platformSanityChecks() {
func PlatformSanityChecks() {
// Nothing to do yet.
}

2
setup/base/sanity_unix.go
View file

@ -9,7 +9,7 @@ import (
"github.com/sirupsen/logrus"
)
func platformSanityChecks() {
func PlatformSanityChecks() {
// Dendrite needs a relatively high number of file descriptors in order
// to function properly, particularly when federating with lots of servers.
// If we run out of file descriptors, we might run into problems accessing

26
setup/monolith.go
View file

@ -21,14 +21,16 @@ import (
"github.com/matrix-org/dendrite/federationapi"
federationAPI "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/internal/transactions"
"github.com/matrix-org/dendrite/mediaapi"
"github.com/matrix-org/dendrite/relayapi"
relayAPI "github.com/matrix-org/dendrite/relayapi/api"
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/setup/process"
"github.com/matrix-org/dendrite/syncapi"
userapi "github.com/matrix-org/dendrite/userapi/api"
"github.com/matrix-org/gomatrixserverlib"
@ -54,23 +56,31 @@ type Monolith struct {
}
// AddAllPublicRoutes attaches all public paths to the given router
func (m *Monolith) AddAllPublicRoutes(base *base.BaseDendrite, natsInstance *jetstream.NATSInstance, caches *caching.Caches) {
func (m *Monolith) AddAllPublicRoutes(
processCtx *process.ProcessContext,
cfg *config.Dendrite,
routers httputil.Routers,
cm sqlutil.Connections,
natsInstance *jetstream.NATSInstance,
caches *caching.Caches,
enableMetrics bool,
) {
userDirectoryProvider := m.ExtUserDirectoryProvider
if userDirectoryProvider == nil {
userDirectoryProvider = m.UserAPI
}
clientapi.AddPublicRoutes(
base.ProcessContext, base.Routers, base.Cfg, natsInstance, m.FedClient, m.RoomserverAPI, m.AppserviceAPI, transactions.New(),
processCtx, routers, cfg, natsInstance, m.FedClient, m.RoomserverAPI, m.AppserviceAPI, transactions.New(),
m.FederationAPI, m.UserAPI, userDirectoryProvider,
m.ExtPublicRoomsProvider, base.EnableMetrics,
m.ExtPublicRoomsProvider, enableMetrics,
)
federationapi.AddPublicRoutes(
base.ProcessContext, base.Routers, base.Cfg, natsInstance, m.UserAPI, m.FedClient, m.KeyRing, m.RoomserverAPI, m.FederationAPI, nil, base.EnableMetrics,
processCtx, routers, cfg, natsInstance, m.UserAPI, m.FedClient, m.KeyRing, m.RoomserverAPI, m.FederationAPI, nil, enableMetrics,
)
mediaapi.AddPublicRoutes(base.Routers.Media, base.ConnectionManager, base.Cfg, m.UserAPI, m.Client)
syncapi.AddPublicRoutes(base.ProcessContext, base.Routers, base.Cfg, base.ConnectionManager, natsInstance, m.UserAPI, m.RoomserverAPI, caches, base.EnableMetrics)
mediaapi.AddPublicRoutes(routers.Media, cm, cfg, m.UserAPI, m.Client)
syncapi.AddPublicRoutes(processCtx, routers, cfg, cm, natsInstance, m.UserAPI, m.RoomserverAPI, caches, enableMetrics)
if m.RelayAPI != nil {
relayapi.AddPublicRoutes(base.Routers, base.Cfg, m.KeyRing, m.RelayAPI)
relayapi.AddPublicRoutes(routers, cfg, m.KeyRing, m.RelayAPI)
}
}

13
setup/mscs/msc2836/msc2836.go
View file

@ -31,8 +31,9 @@ import (
fs "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/internal/hooks"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/internal/sqlutil"
roomserver "github.com/matrix-org/dendrite/roomserver/api"
"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"
"github.com/matrix-org/util"
@ -99,10 +100,10 @@ func toClientResponse(res *MSC2836EventRelationshipsResponse) *EventRelationship
// Enable this MSC
func Enable(
base *base.BaseDendrite, rsAPI roomserver.RoomserverInternalAPI, fsAPI fs.FederationInternalAPI,
cfg *config.Dendrite, cm sqlutil.Connections, routers httputil.Routers, rsAPI roomserver.RoomserverInternalAPI, fsAPI fs.FederationInternalAPI,
userAPI userapi.UserInternalAPI, keyRing gomatrixserverlib.JSONVerifier,
) error {
db, err := NewDatabase(base.ConnectionManager, &base.Cfg.MSCs.Database)
db, err := NewDatabase(cm, &cfg.MSCs.Database)
if err != nil {
return fmt.Errorf("cannot enable MSC2836: %w", err)
}
@ -125,14 +126,14 @@ func Enable(
}
})
base.Routers.Client.Handle("/unstable/event_relationships",
routers.Client.Handle("/unstable/event_relationships",
httputil.MakeAuthAPI("eventRelationships", userAPI, eventRelationshipHandler(db, rsAPI, fsAPI)),
).Methods(http.MethodPost, http.MethodOptions)
base.Routers.Federation.Handle("/unstable/event_relationships", httputil.MakeExternalAPI(
routers.Federation.Handle("/unstable/event_relationships", httputil.MakeExternalAPI(
"msc2836_event_relationships", func(req *http.Request) util.JSONResponse {
fedReq, errResp := gomatrixserverlib.VerifyHTTPRequest(
req, time.Now(), base.Cfg.Global.ServerName, base.Cfg.Global.IsLocalServerName, keyRing,
req, time.Now(), cfg.Global.ServerName, cfg.Global.IsLocalServerName, keyRing,
)
if fedReq == nil {
return errResp

15
setup/mscs/msc2836/msc2836_test.go
View file

@ -15,13 +15,13 @@ import (
"time"
"github.com/gorilla/mux"
"github.com/matrix-org/dendrite/setup/process"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/dendrite/internal/hooks"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/internal/sqlutil"
roomserver "github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/mscs/msc2836"
userapi "github.com/matrix-org/dendrite/userapi/api"
@ -555,21 +555,18 @@ func injectEvents(t *testing.T, userAPI userapi.UserInternalAPI, rsAPI roomserve
cfg.Global.ServerName = "localhost"
cfg.MSCs.Database.ConnectionString = "file:msc2836_test.db"
cfg.MSCs.MSCs = []string{"msc2836"}
cm := sqlutil.NewConnectionManager(nil, config.DatabaseOptions{})
base := &base.BaseDendrite{
Cfg: cfg,
Routers: httputil.NewRouters(),
ConnectionManager: cm,
}
err := msc2836.Enable(base, rsAPI, nil, userAPI, nil)
processCtx := process.NewProcessContext()
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
routers := httputil.NewRouters()
err := msc2836.Enable(cfg, cm, routers, rsAPI, nil, userAPI, nil)
if err != nil {
t.Fatalf("failed to enable MSC2836: %s", err)
}
for _, ev := range events {
hooks.Run(hooks.KindNewEventPersisted, ev)
}
return base.Routers.Client
return routers.Client
}
type fledglingEvent struct {

18
setup/mscs/msc2946/msc2946.go
View file

@ -33,7 +33,7 @@ import (
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/httputil"
roomserver "github.com/matrix-org/dendrite/roomserver/api"
"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"
"github.com/matrix-org/util"
@ -54,17 +54,17 @@ type MSC2946ClientResponse struct {
// Enable this MSC
func Enable(
base *base.BaseDendrite, rsAPI roomserver.RoomserverInternalAPI, userAPI userapi.UserInternalAPI,
cfg *config.Dendrite, routers httputil.Routers, rsAPI roomserver.RoomserverInternalAPI, userAPI userapi.UserInternalAPI,
fsAPI fs.FederationInternalAPI, keyRing gomatrixserverlib.JSONVerifier, cache caching.SpaceSummaryRoomsCache,
) error {
clientAPI := httputil.MakeAuthAPI("spaces", userAPI, spacesHandler(rsAPI, fsAPI, cache, base.Cfg.Global.ServerName), httputil.WithAllowGuests())
base.Routers.Client.Handle("/v1/rooms/{roomID}/hierarchy", clientAPI).Methods(http.MethodGet, http.MethodOptions)
base.Routers.Client.Handle("/unstable/org.matrix.msc2946/rooms/{roomID}/hierarchy", clientAPI).Methods(http.MethodGet, http.MethodOptions)
clientAPI := httputil.MakeAuthAPI("spaces", userAPI, spacesHandler(rsAPI, fsAPI, cache, cfg.Global.ServerName), httputil.WithAllowGuests())
routers.Client.Handle("/v1/rooms/{roomID}/hierarchy", clientAPI).Methods(http.MethodGet, http.MethodOptions)
routers.Client.Handle("/unstable/org.matrix.msc2946/rooms/{roomID}/hierarchy", clientAPI).Methods(http.MethodGet, http.MethodOptions)
fedAPI := httputil.MakeExternalAPI(
"msc2946_fed_spaces", func(req *http.Request) util.JSONResponse {
fedReq, errResp := gomatrixserverlib.VerifyHTTPRequest(
req, time.Now(), base.Cfg.Global.ServerName, base.Cfg.Global.IsLocalServerName, keyRing,
req, time.Now(), cfg.Global.ServerName, cfg.Global.IsLocalServerName, keyRing,
)
if fedReq == nil {
return errResp
@ -75,11 +75,11 @@ func Enable(
return util.ErrorResponse(err)
}
roomID := params["roomID"]
return federatedSpacesHandler(req.Context(), fedReq, roomID, cache, rsAPI, fsAPI, base.Cfg.Global.ServerName)
return federatedSpacesHandler(req.Context(), fedReq, roomID, cache, rsAPI, fsAPI, cfg.Global.ServerName)
},
)
base.Routers.Federation.Handle("/unstable/org.matrix.msc2946/hierarchy/{roomID}", fedAPI).Methods(http.MethodGet)
base.Routers.Federation.Handle("/v1/hierarchy/{roomID}", fedAPI).Methods(http.MethodGet)
routers.Federation.Handle("/unstable/org.matrix.msc2946/hierarchy/{roomID}", fedAPI).Methods(http.MethodGet)
routers.Federation.Handle("/v1/hierarchy/{roomID}", fedAPI).Methods(http.MethodGet)
return nil
}

16
setup/mscs/mscs.go
View file

@ -20,30 +20,32 @@ import (
"fmt"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/httputil"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/setup"
"github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/mscs/msc2836"
"github.com/matrix-org/dendrite/setup/mscs/msc2946"
"github.com/matrix-org/util"
)
// Enable MSCs - returns an error on unknown MSCs
func Enable(base *base.BaseDendrite, monolith *setup.Monolith, caches *caching.Caches) error {
for _, msc := range base.Cfg.MSCs.MSCs {
func Enable(cfg *config.Dendrite, cm sqlutil.Connections, routers httputil.Routers, monolith *setup.Monolith, caches *caching.Caches) error {
for _, msc := range cfg.MSCs.MSCs {
util.GetLogger(context.Background()).WithField("msc", msc).Info("Enabling MSC")
if err := EnableMSC(base, monolith, msc, caches); err != nil {
if err := EnableMSC(cfg, cm, routers, monolith, msc, caches); err != nil {
return err
}
}
return nil
}
func EnableMSC(base *base.BaseDendrite, monolith *setup.Monolith, msc string, caches *caching.Caches) error {
func EnableMSC(cfg *config.Dendrite, cm sqlutil.Connections, routers httputil.Routers, monolith *setup.Monolith, msc string, caches *caching.Caches) error {
switch msc {
case "msc2836":
return msc2836.Enable(base, monolith.RoomserverAPI, monolith.FederationAPI, monolith.UserAPI, monolith.KeyRing)
return msc2836.Enable(cfg, cm, routers, monolith.RoomserverAPI, monolith.FederationAPI, monolith.UserAPI, monolith.KeyRing)
case "msc2946":
return msc2946.Enable(base, monolith.RoomserverAPI, monolith.UserAPI, monolith.FederationAPI, monolith.KeyRing, caches)
return msc2946.Enable(cfg, routers, monolith.RoomserverAPI, monolith.UserAPI, monolith.FederationAPI, monolith.KeyRing, caches)
case "msc2444": // enabled inside federationapi
case "msc2753": // enabled inside clientapi
default:

96
test/testrig/base.go
View file

@ -19,87 +19,11 @@ import (
"path/filepath"
"testing"
"github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/jetstream"
"github.com/matrix-org/dendrite/setup/process"
"github.com/matrix-org/dendrite/test"
"github.com/nats-io/nats.go"
)
func CreateBaseDendrite(t *testing.T, dbType test.DBType) (*base.BaseDendrite, func()) {
var cfg config.Dendrite
cfg.Defaults(config.DefaultOpts{
Generate: false,
SingleDatabase: true,
})
cfg.Global.JetStream.InMemory = true
cfg.FederationAPI.KeyPerspectives = nil
switch dbType {
case test.DBTypePostgres:
cfg.Global.Defaults(config.DefaultOpts{ // autogen a signing key
Generate: true,
SingleDatabase: true,
})
cfg.MediaAPI.Defaults(config.DefaultOpts{ // autogen a media path
Generate: true,
SingleDatabase: true,
})
cfg.SyncAPI.Fulltext.Defaults(config.DefaultOpts{ // use in memory fts
Generate: true,
SingleDatabase: true,
})
cfg.Global.ServerName = "test"
// use a distinct prefix else concurrent postgres/sqlite runs will clash since NATS will use
// the file system event with InMemory=true :(
cfg.Global.JetStream.TopicPrefix = fmt.Sprintf("Test_%d_", dbType)
connStr, close := test.PrepareDBConnectionString(t, dbType)
cfg.Global.DatabaseOptions = config.DatabaseOptions{
ConnectionString: config.DataSource(connStr),
MaxOpenConnections: 10,
MaxIdleConnections: 2,
ConnMaxLifetimeSeconds: 60,
}
base := base.NewBaseDendrite(&cfg, base.DisableMetrics)
return base, func() {
base.ShutdownDendrite()
base.WaitForShutdown()
close()
}
case test.DBTypeSQLite:
cfg.Defaults(config.DefaultOpts{
Generate: true,
SingleDatabase: false,
})
cfg.Global.ServerName = "test"
// use a distinct prefix else concurrent postgres/sqlite runs will clash since NATS will use
// the file system event with InMemory=true :(
cfg.Global.JetStream.TopicPrefix = fmt.Sprintf("Test_%d_", dbType)
// Use a temp dir provided by go for tests, this will be cleanup by a call to t.CleanUp()
tempDir := t.TempDir()
cfg.FederationAPI.Database.ConnectionString = config.DataSource(filepath.Join("file://", tempDir, "federationapi.db"))
cfg.KeyServer.Database.ConnectionString = config.DataSource(filepath.Join("file://", tempDir, "keyserver.db"))
cfg.MSCs.Database.ConnectionString = config.DataSource(filepath.Join("file://", tempDir, "mscs.db"))
cfg.MediaAPI.Database.ConnectionString = config.DataSource(filepath.Join("file://", tempDir, "mediaapi.db"))
cfg.RoomServer.Database.ConnectionString = config.DataSource(filepath.Join("file://", tempDir, "roomserver.db"))
cfg.SyncAPI.Database.ConnectionString = config.DataSource(filepath.Join("file://", tempDir, "syncapi.db"))
cfg.UserAPI.AccountDatabase.ConnectionString = config.DataSource(filepath.Join("file://", tempDir, "userapi.db"))
cfg.RelayAPI.Database.ConnectionString = config.DataSource(filepath.Join("file://", tempDir, "relayapi.db"))
base := base.NewBaseDendrite(&cfg, base.DisableMetrics)
return base, func() {
base.ShutdownDendrite()
base.WaitForShutdown()
t.Cleanup(func() {}) // removes t.TempDir, where all database files are created
}
default:
t.Fatalf("unknown db type: %v", dbType)
}
return nil, nil
}
func CreateConfig(t *testing.T, dbType test.DBType) (*config.Dendrite, *process.ProcessContext, func()) {
var cfg config.Dendrite
cfg.Defaults(config.DefaultOpts{
@ -136,9 +60,9 @@ func CreateConfig(t *testing.T, dbType test.DBType) (*config.Dendrite, *process.
ConnMaxLifetimeSeconds: 60,
}
return &cfg, ctx, func() {
closeDb()
ctx.ShutdownDendrite()
ctx.WaitForShutdown()
closeDb()
}
case test.DBTypeSQLite:
cfg.Defaults(config.DefaultOpts{
@ -165,26 +89,10 @@ func CreateConfig(t *testing.T, dbType test.DBType) (*config.Dendrite, *process.
return &cfg, ctx, func() {
ctx.ShutdownDendrite()
ctx.WaitForShutdown()
t.Cleanup(func() {}) // removes t.TempDir, where all database files are created
}
default:
t.Fatalf("unknown db type: %v", dbType)
}
return &config.Dendrite{}, nil, func() {}
}
func Base(cfg *config.Dendrite) (*base.BaseDendrite, nats.JetStreamContext, *nats.Conn) {
if cfg == nil {
cfg = &config.Dendrite{}
cfg.Defaults(config.DefaultOpts{
Generate: true,
SingleDatabase: false,
})
}
cfg.Global.JetStream.InMemory = true
cfg.SyncAPI.Fulltext.InMemory = true
cfg.FederationAPI.KeyPerspectives = nil
base := base.NewBaseDendrite(cfg, base.DisableMetrics)
natsInstance := jetstream.NATSInstance{}
js, jc := natsInstance.Prepare(base.ProcessContext, &cfg.Global.JetStream)
return base, js, jc
}

14
userapi/util/phonehomestats_test.go
View file

@ -7,10 +7,10 @@ import (
"testing"
"time"
"github.com/matrix-org/dendrite/internal/sqlutil"
"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"
@ -18,12 +18,10 @@ import (
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)
cfg, processCtx, closeDB := testrig.CreateConfig(t, dbType)
defer closeDB()
db, err := storage.NewUserDatabase(b.Context(), b.ConnectionManager, &config.DatabaseOptions{
ConnectionString: config.DataSource(connStr),
}, "localhost", bcrypt.MinCost, 1000, 1000, "")
cm := sqlutil.NewConnectionManager(processCtx, cfg.Global.DatabaseOptions)
db, err := storage.NewUserDatabase(processCtx.Context(), cm, &cfg.UserAPI.AccountDatabase, "localhost", bcrypt.MinCost, 1000, 1000, "")
if err != nil {
t.Error(err)
}
@ -62,12 +60,12 @@ func TestCollect(t *testing.T) {
}))
defer srv.Close()
b.Cfg.Global.ReportStats.Endpoint = srv.URL
cfg.Global.ReportStats.Endpoint = srv.URL
stats := phoneHomeStats{
prevData: timestampToRUUsage{},
serverName: "localhost",
startTime: time.Now(),
cfg: b.Cfg,
cfg: cfg,
db: db,
isMonolith: false,
client: &http.Client{Timeout: time.Second},