diff --git a/build/gobind-pinecone/monolith.go b/build/gobind-pinecone/monolith.go index e30057ed4..000039cc0 100644 --- a/build/gobind-pinecone/monolith.go +++ b/build/gobind-pinecone/monolith.go @@ -264,8 +264,7 @@ func (m *DendriteMonolith) Start() { cfg.Global.ServerName = gomatrixserverlib.ServerName(hex.EncodeToString(pk)) cfg.Global.PrivateKey = sk cfg.Global.KeyID = gomatrixserverlib.KeyID(signing.KeyID) - cfg.Global.Kafka.UseNaffka = true - cfg.Global.Kafka.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/%s-naffka.db", m.StorageDirectory, prefix)) + cfg.Global.JetStream.StoragePath = config.Path(fmt.Sprintf("file:%s/%s", m.StorageDirectory, prefix)) cfg.UserAPI.AccountDatabase.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/%s-account.db", m.StorageDirectory, prefix)) cfg.UserAPI.DeviceDatabase.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/%s-device.db", m.StorageDirectory, prefix)) cfg.MediaAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/%s-mediaapi.db", m.CacheDirectory, prefix)) diff --git a/build/gobind-yggdrasil/monolith.go b/build/gobind-yggdrasil/monolith.go index 338628049..748c35f7b 100644 --- a/build/gobind-yggdrasil/monolith.go +++ b/build/gobind-yggdrasil/monolith.go @@ -89,8 +89,7 @@ func (m *DendriteMonolith) Start() { cfg.Global.ServerName = gomatrixserverlib.ServerName(ygg.DerivedServerName()) cfg.Global.PrivateKey = ygg.SigningPrivateKey() cfg.Global.KeyID = gomatrixserverlib.KeyID(signing.KeyID) - cfg.Global.Kafka.UseNaffka = true - cfg.Global.Kafka.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/dendrite-p2p-naffka.db", m.StorageDirectory)) + cfg.Global.JetStream.StoragePath = config.Path(fmt.Sprintf("file:%s/", m.StorageDirectory)) cfg.UserAPI.AccountDatabase.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/dendrite-p2p-account.db", m.StorageDirectory)) cfg.UserAPI.DeviceDatabase.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/dendrite-p2p-device.db", m.StorageDirectory)) cfg.MediaAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s/dendrite-p2p-mediaapi.db", m.StorageDirectory)) diff --git a/cmd/dendrite-demo-libp2p/main.go b/cmd/dendrite-demo-libp2p/main.go index dc9ff8601..8ebe9541c 100644 --- a/cmd/dendrite-demo-libp2p/main.go +++ b/cmd/dendrite-demo-libp2p/main.go @@ -123,6 +123,7 @@ func main() { cfg.Global.PrivateKey = privKey cfg.Global.KeyID = gomatrixserverlib.KeyID(fmt.Sprintf("ed25519:%s", *instanceName)) cfg.FederationSender.FederationMaxRetries = 6 + cfg.Global.JetStream.StoragePath = config.Path(fmt.Sprintf("%s/", *instanceName)) cfg.UserAPI.AccountDatabase.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-account.db", *instanceName)) cfg.UserAPI.DeviceDatabase.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-device.db", *instanceName)) cfg.MediaAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-mediaapi.db", *instanceName)) diff --git a/cmd/dendrite-demo-pinecone/main.go b/cmd/dendrite-demo-pinecone/main.go index 5ee455eb1..e6a88fcc9 100644 --- a/cmd/dendrite-demo-pinecone/main.go +++ b/cmd/dendrite-demo-pinecone/main.go @@ -144,6 +144,7 @@ func main() { cfg.Global.ServerName = gomatrixserverlib.ServerName(hex.EncodeToString(pk)) cfg.Global.PrivateKey = sk cfg.Global.KeyID = gomatrixserverlib.KeyID(signing.KeyID) + cfg.Global.JetStream.StoragePath = config.Path(fmt.Sprintf("%s/", *instanceName)) cfg.UserAPI.AccountDatabase.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-account.db", *instanceName)) cfg.UserAPI.DeviceDatabase.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-device.db", *instanceName)) cfg.MediaAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-mediaapi.db", *instanceName)) diff --git a/cmd/dendrite-demo-yggdrasil/main.go b/cmd/dendrite-demo-yggdrasil/main.go index 2cb0a7dff..64b186af9 100644 --- a/cmd/dendrite-demo-yggdrasil/main.go +++ b/cmd/dendrite-demo-yggdrasil/main.go @@ -72,6 +72,7 @@ func main() { cfg.Global.ServerName = gomatrixserverlib.ServerName(ygg.DerivedServerName()) cfg.Global.PrivateKey = ygg.SigningPrivateKey() cfg.Global.KeyID = gomatrixserverlib.KeyID(signing.KeyID) + cfg.Global.JetStream.StoragePath = config.Path(fmt.Sprintf("%s/", *instanceName)) cfg.UserAPI.AccountDatabase.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-account.db", *instanceName)) cfg.UserAPI.DeviceDatabase.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-device.db", *instanceName)) cfg.MediaAPI.Database.ConnectionString = config.DataSource(fmt.Sprintf("file:%s-mediaapi.db", *instanceName)) diff --git a/cmd/dendritejs-pinecone/main.go b/cmd/dendritejs-pinecone/main.go index 25e496909..cfefb2a41 100644 --- a/cmd/dendritejs-pinecone/main.go +++ b/cmd/dendritejs-pinecone/main.go @@ -162,8 +162,7 @@ func main() { cfg.SigningKeyServer.Database.ConnectionString = "file:/idb/dendritejs_signingkeyserver.db" cfg.SyncAPI.Database.ConnectionString = "file:/idb/dendritejs_syncapi.db" cfg.KeyServer.Database.ConnectionString = "file:/idb/dendritejs_e2ekey.db" - cfg.Global.Kafka.UseNaffka = true - cfg.Global.Kafka.Database.ConnectionString = "file:/idb/dendritejs_naffka.db" + cfg.Global.Kafka.Database.StorageDirectory = "file:/idb/dendritejs/" cfg.Global.TrustedIDServers = []string{} cfg.Global.KeyID = gomatrixserverlib.KeyID(signing.KeyID) cfg.Global.PrivateKey = sk diff --git a/cmd/dendritejs/main.go b/cmd/dendritejs/main.go index d5a845ae0..dfb2b4719 100644 --- a/cmd/dendritejs/main.go +++ b/cmd/dendritejs/main.go @@ -173,8 +173,7 @@ func main() { cfg.SigningKeyServer.Database.ConnectionString = "file:/idb/dendritejs_signingkeyserver.db" cfg.SyncAPI.Database.ConnectionString = "file:/idb/dendritejs_syncapi.db" cfg.KeyServer.Database.ConnectionString = "file:/idb/dendritejs_e2ekey.db" - cfg.Global.Kafka.UseNaffka = true - cfg.Global.Kafka.Database.ConnectionString = "file:/idb/dendritejs_naffka.db" + cfg.Global.Kafka.Database.StorageDirectory = "file:/idb/dendritejs/" cfg.Global.TrustedIDServers = []string{ "matrix.org", "vector.im", } diff --git a/federationapi/federationapi_test.go b/federationapi/federationapi_test.go index 505a11dae..3d22eb07b 100644 --- a/federationapi/federationapi_test.go +++ b/federationapi/federationapi_test.go @@ -23,8 +23,7 @@ func TestRoomsV3URLEscapeDoNot404(t *testing.T) { cfg.Global.KeyID = gomatrixserverlib.KeyID("ed25519:auto") cfg.Global.ServerName = gomatrixserverlib.ServerName("localhost") cfg.Global.PrivateKey = privKey - cfg.Global.Kafka.UseNaffka = true - cfg.Global.Kafka.Database.ConnectionString = config.DataSource("file::memory:") + cfg.Global.JetStream.InMemory = true cfg.FederationSender.Database.ConnectionString = config.DataSource("file::memory:") base := setup.NewBaseDendrite(cfg, "Monolith", false) keyRing := &test.NopJSONVerifier{} diff --git a/internal/test/config.go b/internal/test/config.go index 7e68d6d2e..b023426e2 100644 --- a/internal/test/config.go +++ b/internal/test/config.go @@ -81,7 +81,7 @@ func MakeConfig(configDir, kafkaURI, database, host string, startPort int) (*con cfg.MediaAPI.BasePath = config.Path(mediaBasePath) - cfg.Global.Kafka.Addresses = []string{kafkaURI} + cfg.Global.JetStream.Addresses = []string{kafkaURI} // TODO: Use different databases for the different schemas. // Using the same database for every schema currently works because diff --git a/setup/config/config.go b/setup/config/config.go index 52307ee5f..0589b5a30 100644 --- a/setup/config/config.go +++ b/setup/config/config.go @@ -296,6 +296,8 @@ func (config *Dendrite) Derive() error { func (c *Dendrite) Defaults() { c.Version = 1 + c.Wiring() + c.Global.Defaults() c.ClientAPI.Defaults() c.EDUServer.Defaults() @@ -309,8 +311,6 @@ func (c *Dendrite) Defaults() { c.UserAPI.Defaults() c.AppServiceAPI.Defaults() c.MSCs.Defaults() - - c.Wiring() } func (c *Dendrite) Verify(configErrs *ConfigErrors, isMonolith bool) { diff --git a/setup/config/config_jetstream.go b/setup/config/config_jetstream.go index 53b1e163c..0f348aa9f 100644 --- a/setup/config/config_jetstream.go +++ b/setup/config/config_jetstream.go @@ -5,21 +5,26 @@ import "fmt" type JetStream struct { Matrix *Global `yaml:"-"` + // Persistent directory to store JetStream streams in. + StoragePath Path `yaml:"storage_path"` // A list of NATS addresses to connect to. If none are specified, an // internal NATS server will be used when running in monolith mode only. Addresses []string `yaml:"addresses"` // The prefix to use for stream names for this homeserver - really only // useful if running more than one Dendrite on the same NATS deployment. TopicPrefix string `yaml:"topic_prefix"` + // Keep all storage in memory. This is mostly useful for unit tests. + InMemory bool `yaml:"-"` } -func (k *JetStream) TopicFor(name string) string { - return fmt.Sprintf("%s%s", k.TopicPrefix, name) +func (c *JetStream) TopicFor(name string) string { + return fmt.Sprintf("%s%s", c.TopicPrefix, name) } func (c *JetStream) Defaults() { c.Addresses = []string{} c.TopicPrefix = "Dendrite" + c.StoragePath = Path(fmt.Sprintf("./%s", c.Matrix.ServerName)) } func (c *JetStream) Verify(configErrs *ConfigErrors, isMonolith bool) { diff --git a/setup/jetstream/nats.go b/setup/jetstream/nats.go index 4127129a8..0300f076d 100644 --- a/setup/jetstream/nats.go +++ b/setup/jetstream/nats.go @@ -27,7 +27,7 @@ func SetupConsumerProducer(cfg *config.JetStream) (sarama.Consumer, sarama.SyncP ServerName: "monolith", DontListen: true, JetStream: true, - StoreDir: string(cfg.Matrix.ServerName), + StoreDir: string(cfg.Matrix.JetStream.StoragePath), }) if err != nil { panic(err) @@ -76,6 +76,12 @@ func setupNATS(cfg *config.JetStream, nc *natsclient.Conn) (sarama.Consumer, sar stream.Name = cfg.TopicFor(stream.Name) stream.Subjects = []string{stream.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 { logrus.WithError(err).WithField("stream", stream.Name).Fatal("Unable to add stream") }