diff --git a/appservice/consumers/roomserver.go b/appservice/consumers/roomserver.go index 139b5724..8aea5c34 100644 --- a/appservice/consumers/roomserver.go +++ b/appservice/consumers/roomserver.go @@ -34,6 +34,7 @@ import ( type OutputRoomEventConsumer struct { ctx context.Context jetstream nats.JetStreamContext + durable nats.SubOpt topic string asDB storage.Database rsAPI api.RoomserverInternalAPI @@ -54,6 +55,7 @@ func NewOutputRoomEventConsumer( return &OutputRoomEventConsumer{ ctx: process.Context(), jetstream: js, + durable: cfg.Global.JetStream.Durable("AppserviceRoomserverConsumer"), topic: cfg.Global.JetStream.TopicFor(jetstream.OutputRoomEvent), asDB: appserviceDB, rsAPI: rsAPI, @@ -64,7 +66,7 @@ func NewOutputRoomEventConsumer( // Start consuming from room servers func (s *OutputRoomEventConsumer) Start() error { - _, err := s.jetstream.Subscribe(s.topic, s.onMessage) + _, err := s.jetstream.Subscribe(s.topic, s.onMessage, s.durable) return err } diff --git a/cmd/dendrite-monolith-server/main.go b/cmd/dendrite-monolith-server/main.go index 08851734..4d0598f3 100644 --- a/cmd/dendrite-monolith-server/main.go +++ b/cmd/dendrite-monolith-server/main.go @@ -99,10 +99,6 @@ func main() { } keyRing := fsAPI.KeyRing() - // 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 dependency - rsImpl.SetFederationAPI(fsAPI, keyRing) - keyImpl := keyserver.NewInternalAPI(base, &base.Cfg.KeyServer, fsAPI) keyAPI := keyImpl if base.UseHTTPAPIs { diff --git a/federationapi/consumers/eduserver.go b/federationapi/consumers/eduserver.go index 9e52acef..c3e5b4d4 100644 --- a/federationapi/consumers/eduserver.go +++ b/federationapi/consumers/eduserver.go @@ -34,6 +34,7 @@ import ( type OutputEDUConsumer struct { ctx context.Context jetstream nats.JetStreamContext + durable nats.SubOpt db storage.Database queues *queue.OutgoingQueues ServerName gomatrixserverlib.ServerName @@ -56,6 +57,7 @@ func NewOutputEDUConsumer( queues: queues, db: store, ServerName: cfg.Matrix.ServerName, + durable: cfg.Matrix.JetStream.Durable("FederationAPIEDUServerConsumer"), typingTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputTypingEvent), sendToDeviceTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputSendToDeviceEvent), receiptTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputReceiptEvent), @@ -64,13 +66,13 @@ func NewOutputEDUConsumer( // Start consuming from EDU servers func (t *OutputEDUConsumer) Start() error { - if _, err := t.jetstream.Subscribe(t.typingTopic, t.onTypingEvent); err != nil { + if _, err := t.jetstream.Subscribe(t.typingTopic, t.onTypingEvent, t.durable); err != nil { return err } - if _, err := t.jetstream.Subscribe(t.sendToDeviceTopic, t.onSendToDeviceEvent); err != nil { + if _, err := t.jetstream.Subscribe(t.sendToDeviceTopic, t.onSendToDeviceEvent, t.durable); err != nil { return err } - if _, err := t.jetstream.Subscribe(t.receiptTopic, t.onReceiptEvent); err != nil { + if _, err := t.jetstream.Subscribe(t.receiptTopic, t.onReceiptEvent, t.durable); err != nil { return err } return nil diff --git a/federationapi/consumers/roomserver.go b/federationapi/consumers/roomserver.go index 12410bb7..632adae3 100644 --- a/federationapi/consumers/roomserver.go +++ b/federationapi/consumers/roomserver.go @@ -37,6 +37,7 @@ type OutputRoomEventConsumer struct { cfg *config.FederationAPI rsAPI api.RoomserverInternalAPI jetstream nats.JetStreamContext + durable nats.SubOpt db storage.Database queues *queue.OutgoingQueues topic string @@ -58,13 +59,14 @@ func NewOutputRoomEventConsumer( db: store, queues: queues, rsAPI: rsAPI, + durable: cfg.Matrix.JetStream.Durable("FederationAPIRoomServerConsumer"), topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputRoomEvent), } } // Start consuming from room servers func (s *OutputRoomEventConsumer) Start() error { - _, err := s.jetstream.Subscribe(s.topic, s.onMessage) + _, err := s.jetstream.Subscribe(s.topic, s.onMessage, s.durable) return err } diff --git a/federationapi/federationapi_keys_test.go b/federationapi/federationapi_keys_test.go index b9503963..4774c882 100644 --- a/federationapi/federationapi_keys_test.go +++ b/federationapi/federationapi_keys_test.go @@ -68,6 +68,13 @@ func TestMain(m *testing.M) { panic("can't create cache: " + err.Error()) } + // Create a temporary directory for JetStream. + d, err := ioutil.TempDir("./", "jetstream*") + if err != nil { + panic(err) + } + defer os.RemoveAll(d) + // Draw up just enough Dendrite config for the server key // API to work. cfg := &config.Dendrite{} @@ -75,6 +82,8 @@ func TestMain(m *testing.M) { cfg.Global.ServerName = gomatrixserverlib.ServerName(s.name) cfg.Global.PrivateKey = testPriv cfg.Global.JetStream.InMemory = true + cfg.Global.JetStream.TopicPrefix = string(s.name[:1]) + cfg.Global.JetStream.StoragePath = config.Path(d) cfg.Global.KeyID = serverKeyID cfg.Global.KeyValidityPeriod = s.validity cfg.FederationAPI.Database.ConnectionString = config.DataSource("file::memory:") diff --git a/roomserver/internal/api.go b/roomserver/internal/api.go index 859c74b2..7a116e30 100644 --- a/roomserver/internal/api.go +++ b/roomserver/internal/api.go @@ -73,6 +73,7 @@ func NewRoomserverAPI( InputRoomEventTopic: inputRoomEventTopic, OutputRoomEventTopic: outputRoomEventTopic, JetStream: consumer, + Durable: cfg.Matrix.JetStream.Durable("RoomserverInputConsumer"), ServerName: cfg.Matrix.ServerName, ACLs: serverACLs, }, diff --git a/roomserver/internal/input/input.go b/roomserver/internal/input/input.go index a124b677..bf816f41 100644 --- a/roomserver/internal/input/input.go +++ b/roomserver/internal/input/input.go @@ -45,6 +45,7 @@ var keyContentFields = map[string]string{ type Inputer struct { DB storage.Database JetStream nats.JetStreamContext + Durable nats.SubOpt ServerName gomatrixserverlib.ServerName FSAPI fedapi.FederationInternalAPI KeyRing gomatrixserverlib.JSONVerifier @@ -91,6 +92,8 @@ func (r *Inputer) Start() error { // or nak them within a certain amount of time. This stops that from // happening, so we don't end up doing a lot of unnecessary duplicate work. nats.MaxDeliver(0), + // Use a durable named consumer. + r.Durable, ) return err } diff --git a/setup/config/config_jetstream.go b/setup/config/config_jetstream.go index 0bd84899..94e2d88b 100644 --- a/setup/config/config_jetstream.go +++ b/setup/config/config_jetstream.go @@ -2,6 +2,8 @@ package config import ( "fmt" + + "github.com/nats-io/nats.go" ) type JetStream struct { @@ -23,6 +25,10 @@ func (c *JetStream) TopicFor(name string) string { return fmt.Sprintf("%s%s", c.TopicPrefix, name) } +func (c *JetStream) Durable(name string) nats.SubOpt { + return nats.Durable(c.TopicFor(name)) +} + func (c *JetStream) Defaults(generate bool) { c.Addresses = []string{} c.TopicPrefix = "Dendrite" diff --git a/setup/jetstream/nats.go b/setup/jetstream/nats.go index eec68d82..6dbbd1f4 100644 --- a/setup/jetstream/nats.go +++ b/setup/jetstream/nats.go @@ -75,13 +75,18 @@ func setupNATS(cfg *config.JetStream, nc *natsclient.Conn) (nats.JetStreamContex } if info == nil { stream.Subjects = []string{name} + // If we're trying to keep everything in memory (e.g. unit tests) // then overwrite the storage policy. if cfg.InMemory { stream.Storage = nats.MemoryStorage } - if _, err = s.AddStream(stream); err != nil { + // Namespace the streams without modifying the original streams + // array, otherwise we end up with namespaces on namespaces. + namespaced := *stream + namespaced.Name = name + if _, err = s.AddStream(&namespaced); err != nil { logrus.WithError(err).WithField("stream", name).Fatal("Unable to add stream") } } diff --git a/syncapi/consumers/clientapi.go b/syncapi/consumers/clientapi.go index 85710cdd..1ec9beb0 100644 --- a/syncapi/consumers/clientapi.go +++ b/syncapi/consumers/clientapi.go @@ -34,6 +34,7 @@ import ( type OutputClientDataConsumer struct { ctx context.Context jetstream nats.JetStreamContext + durable nats.SubOpt topic string db storage.Database stream types.StreamProvider @@ -53,6 +54,7 @@ func NewOutputClientDataConsumer( ctx: process.Context(), jetstream: js, topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputClientData), + durable: cfg.Matrix.JetStream.Durable("SyncAPIClientAPIConsumer"), db: store, notifier: notifier, stream: stream, @@ -61,7 +63,7 @@ func NewOutputClientDataConsumer( // Start consuming from room servers func (s *OutputClientDataConsumer) Start() error { - _, err := s.jetstream.Subscribe(s.topic, s.onMessage) + _, err := s.jetstream.Subscribe(s.topic, s.onMessage, s.durable) return err } diff --git a/syncapi/consumers/eduserver_receipts.go b/syncapi/consumers/eduserver_receipts.go index 582e1d64..57d69d6f 100644 --- a/syncapi/consumers/eduserver_receipts.go +++ b/syncapi/consumers/eduserver_receipts.go @@ -34,6 +34,7 @@ import ( type OutputReceiptEventConsumer struct { ctx context.Context jetstream nats.JetStreamContext + durable nats.SubOpt topic string db storage.Database stream types.StreamProvider @@ -54,6 +55,7 @@ func NewOutputReceiptEventConsumer( ctx: process.Context(), jetstream: js, topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputReceiptEvent), + durable: cfg.Matrix.JetStream.Durable("SyncAPIEDUServerReceiptConsumer"), db: store, notifier: notifier, stream: stream, @@ -62,7 +64,7 @@ func NewOutputReceiptEventConsumer( // Start consuming from EDU api func (s *OutputReceiptEventConsumer) Start() error { - _, err := s.jetstream.Subscribe(s.topic, s.onMessage) + _, err := s.jetstream.Subscribe(s.topic, s.onMessage, s.durable) return err } diff --git a/syncapi/consumers/eduserver_sendtodevice.go b/syncapi/consumers/eduserver_sendtodevice.go index 6579c303..54e689fa 100644 --- a/syncapi/consumers/eduserver_sendtodevice.go +++ b/syncapi/consumers/eduserver_sendtodevice.go @@ -36,6 +36,7 @@ import ( type OutputSendToDeviceEventConsumer struct { ctx context.Context jetstream nats.JetStreamContext + durable nats.SubOpt topic string db storage.Database serverName gomatrixserverlib.ServerName // our server name @@ -57,6 +58,7 @@ func NewOutputSendToDeviceEventConsumer( ctx: process.Context(), jetstream: js, topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputSendToDeviceEvent), + durable: cfg.Matrix.JetStream.Durable("SyncAPIEDUServerSendToDeviceConsumer"), db: store, serverName: cfg.Matrix.ServerName, notifier: notifier, @@ -66,7 +68,7 @@ func NewOutputSendToDeviceEventConsumer( // Start consuming from EDU api func (s *OutputSendToDeviceEventConsumer) Start() error { - _, err := s.jetstream.Subscribe(s.topic, s.onMessage) + _, err := s.jetstream.Subscribe(s.topic, s.onMessage, s.durable) return err } diff --git a/syncapi/consumers/eduserver_typing.go b/syncapi/consumers/eduserver_typing.go index 487befe8..de2f6f95 100644 --- a/syncapi/consumers/eduserver_typing.go +++ b/syncapi/consumers/eduserver_typing.go @@ -35,6 +35,7 @@ import ( type OutputTypingEventConsumer struct { ctx context.Context jetstream nats.JetStreamContext + durable nats.SubOpt topic string eduCache *cache.EDUCache stream types.StreamProvider @@ -56,6 +57,7 @@ func NewOutputTypingEventConsumer( ctx: process.Context(), jetstream: js, topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputTypingEvent), + durable: cfg.Matrix.JetStream.Durable("SyncAPIEDUServerTypingConsumer"), eduCache: eduCache, notifier: notifier, stream: stream, @@ -64,7 +66,7 @@ func NewOutputTypingEventConsumer( // Start consuming from EDU api func (s *OutputTypingEventConsumer) Start() error { - _, err := s.jetstream.Subscribe(s.topic, s.onMessage) + _, err := s.jetstream.Subscribe(s.topic, s.onMessage, s.durable) return err } diff --git a/syncapi/consumers/roomserver.go b/syncapi/consumers/roomserver.go index 5b008e3d..6b3ebe53 100644 --- a/syncapi/consumers/roomserver.go +++ b/syncapi/consumers/roomserver.go @@ -38,6 +38,7 @@ type OutputRoomEventConsumer struct { cfg *config.SyncAPI rsAPI api.RoomserverInternalAPI jetstream nats.JetStreamContext + durable nats.SubOpt topic string db storage.Database pduStream types.StreamProvider @@ -61,6 +62,7 @@ func NewOutputRoomEventConsumer( cfg: cfg, jetstream: js, topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputRoomEvent), + durable: cfg.Matrix.JetStream.Durable("SyncAPIRoomServerConsumer"), db: store, notifier: notifier, pduStream: pduStream, @@ -71,7 +73,7 @@ func NewOutputRoomEventConsumer( // Start consuming from room servers func (s *OutputRoomEventConsumer) Start() error { - _, err := s.jetstream.Subscribe(s.topic, s.onMessage) + _, err := s.jetstream.Subscribe(s.topic, s.onMessage, s.durable) return err }