From 9572f5ed19abc0b635092108aa6956eaebc60578 Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Mon, 21 Mar 2022 10:32:34 +0000 Subject: [PATCH] Wait for safe shutdown of NATS Server (#2289) --- appservice/appservice.go | 2 +- clientapi/clientapi.go | 4 +++- .../personalities/clientapi.go | 6 +++--- eduserver/eduserver.go | 2 +- federationapi/federationapi.go | 2 +- keyserver/keyserver.go | 2 +- roomserver/roomserver.go | 2 +- setup/jetstream/nats.go | 14 ++++++++++++-- setup/monolith.go | 2 +- syncapi/syncapi.go | 2 +- userapi/userapi.go | 2 +- 11 files changed, 26 insertions(+), 14 deletions(-) diff --git a/appservice/appservice.go b/appservice/appservice.go index 3e19e09b..b9909186 100644 --- a/appservice/appservice.go +++ b/appservice/appservice.go @@ -59,7 +59,7 @@ func NewInternalAPI( }, }, } - js, _ := jetstream.Prepare(&base.Cfg.Global.JetStream) + js, _ := jetstream.Prepare(base.ProcessContext, &base.Cfg.Global.JetStream) // Create a connection to the appservice postgres DB appserviceDB, err := storage.NewDatabase(&base.Cfg.AppServiceAPI.Database) diff --git a/clientapi/clientapi.go b/clientapi/clientapi.go index e4279c22..75184d3b 100644 --- a/clientapi/clientapi.go +++ b/clientapi/clientapi.go @@ -27,6 +27,7 @@ import ( roomserverAPI "github.com/matrix-org/dendrite/roomserver/api" "github.com/matrix-org/dendrite/setup/config" "github.com/matrix-org/dendrite/setup/jetstream" + "github.com/matrix-org/dendrite/setup/process" userapi "github.com/matrix-org/dendrite/userapi/api" userdb "github.com/matrix-org/dendrite/userapi/storage" "github.com/matrix-org/gomatrixserverlib" @@ -34,6 +35,7 @@ import ( // AddPublicRoutes sets up and registers HTTP handlers for the ClientAPI component. func AddPublicRoutes( + process *process.ProcessContext, router *mux.Router, synapseAdminRouter *mux.Router, cfg *config.ClientAPI, @@ -49,7 +51,7 @@ func AddPublicRoutes( extRoomsProvider api.ExtraPublicRoomsProvider, mscCfg *config.MSCs, ) { - js, _ := jetstream.Prepare(&cfg.Matrix.JetStream) + js, _ := jetstream.Prepare(process, &cfg.Matrix.JetStream) syncProducer := &producers.SyncAPIProducer{ JetStream: js, diff --git a/cmd/dendrite-polylith-multi/personalities/clientapi.go b/cmd/dendrite-polylith-multi/personalities/clientapi.go index bd9f7a10..5e67acd0 100644 --- a/cmd/dendrite-polylith-multi/personalities/clientapi.go +++ b/cmd/dendrite-polylith-multi/personalities/clientapi.go @@ -33,9 +33,9 @@ func ClientAPI(base *basepkg.BaseDendrite, cfg *config.Dendrite) { keyAPI := base.KeyServerHTTPClient() clientapi.AddPublicRoutes( - base.PublicClientAPIMux, base.SynapseAdminMux, &base.Cfg.ClientAPI, accountDB, federation, - rsAPI, eduInputAPI, asQuery, transactions.New(), fsAPI, userAPI, keyAPI, nil, - &cfg.MSCs, + base.ProcessContext, base.PublicClientAPIMux, base.SynapseAdminMux, &base.Cfg.ClientAPI, + accountDB, federation, rsAPI, eduInputAPI, asQuery, transactions.New(), fsAPI, userAPI, + keyAPI, nil, &cfg.MSCs, ) base.SetupAndServeHTTP( diff --git a/eduserver/eduserver.go b/eduserver/eduserver.go index 6882399d..9fe8704c 100644 --- a/eduserver/eduserver.go +++ b/eduserver/eduserver.go @@ -42,7 +42,7 @@ func NewInternalAPI( ) api.EDUServerInputAPI { cfg := &base.Cfg.EDUServer - js, _ := jetstream.Prepare(&cfg.Matrix.JetStream) + js, _ := jetstream.Prepare(base.ProcessContext, &cfg.Matrix.JetStream) return &input.EDUServerInputAPI{ Cache: eduCache, diff --git a/federationapi/federationapi.go b/federationapi/federationapi.go index 9f149d97..b7f93ecb 100644 --- a/federationapi/federationapi.go +++ b/federationapi/federationapi.go @@ -92,7 +92,7 @@ func NewInternalAPI( FailuresUntilBlacklist: cfg.FederationMaxRetries, } - js, _ := jetstream.Prepare(&cfg.Matrix.JetStream) + js, _ := jetstream.Prepare(base.ProcessContext, &cfg.Matrix.JetStream) queues := queue.NewOutgoingQueues( federationDB, base.ProcessContext, diff --git a/keyserver/keyserver.go b/keyserver/keyserver.go index 8a0ce617..cf66bd38 100644 --- a/keyserver/keyserver.go +++ b/keyserver/keyserver.go @@ -39,7 +39,7 @@ func AddInternalRoutes(router *mux.Router, intAPI api.KeyInternalAPI) { func NewInternalAPI( base *base.BaseDendrite, cfg *config.KeyServer, fedClient fedsenderapi.FederationClient, ) api.KeyInternalAPI { - js, _ := jetstream.Prepare(&cfg.Matrix.JetStream) + js, _ := jetstream.Prepare(base.ProcessContext, &cfg.Matrix.JetStream) db, err := storage.NewDatabase(&cfg.Database) if err != nil { diff --git a/roomserver/roomserver.go b/roomserver/roomserver.go index 1992ac33..896773ba 100644 --- a/roomserver/roomserver.go +++ b/roomserver/roomserver.go @@ -50,7 +50,7 @@ func NewInternalAPI( logrus.WithError(err).Panicf("failed to connect to room server db") } - js, nc := jetstream.Prepare(&cfg.Matrix.JetStream) + js, nc := jetstream.Prepare(base.ProcessContext, &cfg.Matrix.JetStream) return internal.NewRoomserverAPI( base.ProcessContext, cfg, roomserverDB, js, nc, diff --git a/setup/jetstream/nats.go b/setup/jetstream/nats.go index 37597d58..43cc0331 100644 --- a/setup/jetstream/nats.go +++ b/setup/jetstream/nats.go @@ -6,6 +6,7 @@ import ( "time" "github.com/matrix-org/dendrite/setup/config" + "github.com/matrix-org/dendrite/setup/process" "github.com/sirupsen/logrus" natsserver "github.com/nats-io/nats-server/v2/server" @@ -15,7 +16,7 @@ import ( var natsServer *natsserver.Server var natsServerMutex sync.Mutex -func Prepare(cfg *config.JetStream) (natsclient.JetStreamContext, *natsclient.Conn) { +func Prepare(process *process.ProcessContext, cfg *config.JetStream) (natsclient.JetStreamContext, *natsclient.Conn) { // check if we need an in-process NATS Server if len(cfg.Addresses) != 0 { return setupNATS(cfg, nil) @@ -35,7 +36,16 @@ func Prepare(cfg *config.JetStream) (natsclient.JetStreamContext, *natsclient.Co panic(err) } natsServer.ConfigureLogger() - go natsServer.Start() + go func() { + process.ComponentStarted() + natsServer.Start() + }() + go func() { + <-process.WaitForShutdown() + natsServer.Shutdown() + natsServer.WaitForShutdown() + process.ComponentFinished() + }() } natsServerMutex.Unlock() if !natsServer.ReadyForConnections(time.Second * 10) { diff --git a/setup/monolith.go b/setup/monolith.go index 7dbd2eea..fa6d962c 100644 --- a/setup/monolith.go +++ b/setup/monolith.go @@ -57,7 +57,7 @@ type Monolith struct { // AddAllPublicRoutes attaches all public paths to the given router func (m *Monolith) AddAllPublicRoutes(process *process.ProcessContext, csMux, ssMux, keyMux, wkMux, mediaMux, synapseMux *mux.Router) { clientapi.AddPublicRoutes( - csMux, synapseMux, &m.Config.ClientAPI, m.AccountDB, + process, csMux, synapseMux, &m.Config.ClientAPI, m.AccountDB, m.FedClient, m.RoomserverAPI, m.EDUInternalAPI, m.AppserviceAPI, transactions.New(), m.FederationAPI, m.UserAPI, m.KeyAPI, diff --git a/syncapi/syncapi.go b/syncapi/syncapi.go index 41635c91..f1f82722 100644 --- a/syncapi/syncapi.go +++ b/syncapi/syncapi.go @@ -49,7 +49,7 @@ func AddPublicRoutes( federation *gomatrixserverlib.FederationClient, cfg *config.SyncAPI, ) { - js, _ := jetstream.Prepare(&cfg.Matrix.JetStream) + js, _ := jetstream.Prepare(process, &cfg.Matrix.JetStream) syncDB, err := storage.NewSyncServerDatasource(&cfg.Database) if err != nil { diff --git a/userapi/userapi.go b/userapi/userapi.go index 1e4ebcb2..97bdf7b2 100644 --- a/userapi/userapi.go +++ b/userapi/userapi.go @@ -46,7 +46,7 @@ func NewInternalAPI( appServices []config.ApplicationService, keyAPI keyapi.KeyInternalAPI, rsAPI rsapi.RoomserverInternalAPI, pgClient pushgateway.Client, ) api.UserInternalAPI { - js, _ := jetstream.Prepare(&cfg.Matrix.JetStream) + js, _ := jetstream.Prepare(base.ProcessContext, &cfg.Matrix.JetStream) syncProducer := producers.NewSyncAPI( db, js,