diff --git a/appservice/appservice.go b/appservice/appservice.go index 4ff42360b..10e4dfd12 100644 --- a/appservice/appservice.go +++ b/appservice/appservice.go @@ -58,7 +58,7 @@ func NewInternalAPI( }, }, } - consumer, _ := jetstream.SetupConsumerProducer(&base.Cfg.Global.JetStream) + js, _, _ := jetstream.Prepare(&base.Cfg.Global.JetStream) // Create a connection to the appservice postgres DB appserviceDB, err := storage.NewDatabase(&base.Cfg.AppServiceAPI.Database) @@ -97,7 +97,7 @@ func NewInternalAPI( // We can't add ASes at runtime so this is safe to do. if len(workerStates) > 0 { consumer := consumers.NewOutputRoomEventConsumer( - base.ProcessContext, base.Cfg, consumer, appserviceDB, + base.ProcessContext, base.Cfg, js, appserviceDB, rsAPI, workerStates, ) if err := consumer.Start(); err != nil { diff --git a/appservice/consumers/roomserver.go b/appservice/consumers/roomserver.go index f9790a0b1..ddcc478c6 100644 --- a/appservice/consumers/roomserver.go +++ b/appservice/consumers/roomserver.go @@ -20,24 +20,24 @@ import ( "github.com/matrix-org/dendrite/appservice/storage" "github.com/matrix-org/dendrite/appservice/types" - "github.com/matrix-org/dendrite/internal" "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" "github.com/matrix-org/gomatrixserverlib" + "github.com/nats-io/nats.go" - "github.com/Shopify/sarama" log "github.com/sirupsen/logrus" ) // OutputRoomEventConsumer consumes events that originated in the room server. type OutputRoomEventConsumer struct { - roomServerConsumer *internal.ContinualConsumer - asDB storage.Database - rsAPI api.RoomserverInternalAPI - serverName string - workerStates []types.ApplicationServiceWorkerState + jetstream nats.JetStreamContext + topic string + asDB storage.Database + rsAPI api.RoomserverInternalAPI + serverName string + workerStates []types.ApplicationServiceWorkerState } // NewOutputRoomEventConsumer creates a new OutputRoomEventConsumer. Call @@ -45,55 +45,54 @@ type OutputRoomEventConsumer struct { func NewOutputRoomEventConsumer( process *process.ProcessContext, cfg *config.Dendrite, - kafkaConsumer sarama.Consumer, + js nats.JetStreamContext, appserviceDB storage.Database, rsAPI api.RoomserverInternalAPI, workerStates []types.ApplicationServiceWorkerState, ) *OutputRoomEventConsumer { - consumer := internal.ContinualConsumer{ - Process: process, - ComponentName: "appservice/roomserver", - Topic: cfg.Global.JetStream.TopicFor(jetstream.OutputRoomEvent), - Consumer: kafkaConsumer, - PartitionStore: appserviceDB, + return &OutputRoomEventConsumer{ + jetstream: js, + topic: cfg.Global.JetStream.TopicFor(jetstream.OutputRoomEvent), + asDB: appserviceDB, + rsAPI: rsAPI, + serverName: string(cfg.Global.ServerName), + workerStates: workerStates, } - s := &OutputRoomEventConsumer{ - roomServerConsumer: &consumer, - asDB: appserviceDB, - rsAPI: rsAPI, - serverName: string(cfg.Global.ServerName), - workerStates: workerStates, - } - consumer.ProcessMessage = s.onMessage - - return s } // Start consuming from room servers func (s *OutputRoomEventConsumer) Start() error { - return s.roomServerConsumer.Start() + _, err := s.jetstream.Subscribe(s.topic, s.onMessage) + return err } // onMessage is called when the appservice component receives a new event from // the room server output log. -func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error { +func (s *OutputRoomEventConsumer) onMessage(msg *nats.Msg) { // Parse out the event JSON var output api.OutputEvent - if err := json.Unmarshal(msg.Value, &output); err != nil { + if err := json.Unmarshal(msg.Data, &output); err != nil { // If the message was invalid, log it and move on to the next message in the stream log.WithError(err).Errorf("roomserver output log: message parse failure") - return nil + _ = msg.Ack() + return } if output.Type != api.OutputTypeNewRoomEvent { - return nil + _ = msg.Ack() + return } events := []*gomatrixserverlib.HeaderedEvent{output.NewRoomEvent.Event} events = append(events, output.NewRoomEvent.AddStateEvents...) // Send event to any relevant application services - return s.filterRoomserverEvents(context.TODO(), events) + if err := s.filterRoomserverEvents(context.TODO(), events); err != nil { + log.WithError(err).Errorf("roomserver output log: filter error") + return + } + + _ = msg.Ack() } // filterRoomserverEvents takes in events and decides whether any of them need diff --git a/appservice/storage/interface.go b/appservice/storage/interface.go index 735e2f90a..25d35af6c 100644 --- a/appservice/storage/interface.go +++ b/appservice/storage/interface.go @@ -17,12 +17,10 @@ package storage import ( "context" - "github.com/matrix-org/dendrite/internal" "github.com/matrix-org/gomatrixserverlib" ) type Database interface { - internal.PartitionStorer StoreEvent(ctx context.Context, appServiceID string, event *gomatrixserverlib.HeaderedEvent) error GetEventsWithAppServiceID(ctx context.Context, appServiceID string, limit int) (int, int, []gomatrixserverlib.HeaderedEvent, bool, error) CountEventsWithAppServiceID(ctx context.Context, appServiceID string) (int, error) diff --git a/clientapi/clientapi.go b/clientapi/clientapi.go index ffab1337d..f8588c83c 100644 --- a/clientapi/clientapi.go +++ b/clientapi/clientapi.go @@ -49,11 +49,11 @@ func AddPublicRoutes( extRoomsProvider api.ExtraPublicRoomsProvider, mscCfg *config.MSCs, ) { - _, producer := jetstream.SetupConsumerProducer(&cfg.Matrix.JetStream) + js, _, _ := jetstream.Prepare(&cfg.Matrix.JetStream) syncProducer := &producers.SyncAPIProducer{ - Producer: producer, - Topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputClientData), + JetStream: js, + Topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputClientData), } routing.Setup( diff --git a/clientapi/producers/syncapi.go b/clientapi/producers/syncapi.go index 6ab8eef28..bd6af5f1f 100644 --- a/clientapi/producers/syncapi.go +++ b/clientapi/producers/syncapi.go @@ -17,39 +17,42 @@ package producers import ( "encoding/json" - "github.com/Shopify/sarama" "github.com/matrix-org/dendrite/internal/eventutil" + "github.com/matrix-org/dendrite/setup/jetstream" + "github.com/nats-io/nats.go" log "github.com/sirupsen/logrus" ) // SyncAPIProducer produces events for the sync API server to consume type SyncAPIProducer struct { - Topic string - Producer sarama.SyncProducer + Topic string + JetStream nats.JetStreamContext } // SendData sends account data to the sync API server func (p *SyncAPIProducer) SendData(userID string, roomID string, dataType string) error { - var m sarama.ProducerMessage + m := &nats.Msg{ + Subject: p.Topic, + Header: nats.Header{}, + } + m.Header.Set(jetstream.UserID, userID) data := eventutil.AccountData{ RoomID: roomID, Type: dataType, } - value, err := json.Marshal(data) + var err error + m.Data, err = json.Marshal(data) if err != nil { return err } - m.Topic = string(p.Topic) - m.Key = sarama.StringEncoder(userID) - m.Value = sarama.ByteEncoder(value) log.WithFields(log.Fields{ "user_id": userID, "room_id": roomID, "data_type": dataType, }).Infof("Producing to topic '%s'", p.Topic) - _, _, err = p.Producer.SendMessage(&m) + _, err = p.JetStream.PublishMsg(m) return err } diff --git a/clientapi/routing/createroom.go b/clientapi/routing/createroom.go index 8f96c3d35..85331192b 100644 --- a/clientapi/routing/createroom.go +++ b/clientapi/routing/createroom.go @@ -463,6 +463,7 @@ func createRoom( }, ev.Headered(roomVersion), nil, + false, ); err != nil { util.GetLogger(req.Context()).WithError(err).Error("SendEventWithState failed") return jsonerror.InternalServerError() diff --git a/clientapi/routing/membership.go b/clientapi/routing/membership.go index 33fb38831..7ddb827eb 100644 --- a/clientapi/routing/membership.go +++ b/clientapi/routing/membership.go @@ -110,6 +110,7 @@ func sendMembership(ctx context.Context, accountDB accounts.Database, device *us []*gomatrixserverlib.HeaderedEvent{event.Event.Headered(roomVer)}, cfg.Matrix.ServerName, nil, + false, ); err != nil { util.GetLogger(ctx).WithError(err).Error("SendEvents failed") return jsonerror.InternalServerError() diff --git a/clientapi/routing/profile.go b/clientapi/routing/profile.go index 7bea35e50..9de1869bf 100644 --- a/clientapi/routing/profile.go +++ b/clientapi/routing/profile.go @@ -169,7 +169,7 @@ func SetAvatarURL( return jsonerror.InternalServerError() } - if err := api.SendEvents(req.Context(), rsAPI, api.KindNew, events, cfg.Matrix.ServerName, nil); err != nil { + if err := api.SendEvents(req.Context(), rsAPI, api.KindNew, events, cfg.Matrix.ServerName, nil, false); err != nil { util.GetLogger(req.Context()).WithError(err).Error("SendEvents failed") return jsonerror.InternalServerError() } @@ -286,7 +286,7 @@ func SetDisplayName( return jsonerror.InternalServerError() } - if err := api.SendEvents(req.Context(), rsAPI, api.KindNew, events, cfg.Matrix.ServerName, nil); err != nil { + if err := api.SendEvents(req.Context(), rsAPI, api.KindNew, events, cfg.Matrix.ServerName, nil, false); err != nil { util.GetLogger(req.Context()).WithError(err).Error("SendEvents failed") return jsonerror.InternalServerError() } diff --git a/clientapi/routing/redaction.go b/clientapi/routing/redaction.go index c25ca4eff..8492236b6 100644 --- a/clientapi/routing/redaction.go +++ b/clientapi/routing/redaction.go @@ -120,7 +120,7 @@ func SendRedaction( JSON: jsonerror.NotFound("Room does not exist"), } } - if err = roomserverAPI.SendEvents(context.Background(), rsAPI, roomserverAPI.KindNew, []*gomatrixserverlib.HeaderedEvent{e}, cfg.Matrix.ServerName, nil); err != nil { + if err = roomserverAPI.SendEvents(context.Background(), rsAPI, roomserverAPI.KindNew, []*gomatrixserverlib.HeaderedEvent{e}, cfg.Matrix.ServerName, nil, false); err != nil { util.GetLogger(req.Context()).WithError(err).Errorf("failed to SendEvents") return jsonerror.InternalServerError() } diff --git a/clientapi/routing/sendevent.go b/clientapi/routing/sendevent.go index 204d2592a..f04983122 100644 --- a/clientapi/routing/sendevent.go +++ b/clientapi/routing/sendevent.go @@ -122,6 +122,7 @@ func SendEvent( }, cfg.Matrix.ServerName, txnAndSessionID, + false, ); err != nil { util.GetLogger(req.Context()).WithError(err).Error("SendEvents failed") return jsonerror.InternalServerError() diff --git a/clientapi/threepid/invites.go b/clientapi/threepid/invites.go index 53cd6b8ca..985cf00c4 100644 --- a/clientapi/threepid/invites.go +++ b/clientapi/threepid/invites.go @@ -367,5 +367,6 @@ func emit3PIDInviteEvent( }, cfg.Matrix.ServerName, nil, + false, ) } diff --git a/dendrite-config.yaml b/dendrite-config.yaml index e53dce8bd..5ed899124 100644 --- a/dendrite-config.yaml +++ b/dendrite-config.yaml @@ -79,7 +79,7 @@ global: in_memory: false # Persistent directory to store JetStream streams in. - storage_path: + storage_path: ./ # The prefix to use for stream names for this homeserver - really only # useful if running more than one Dendrite on the same NATS deployment. @@ -109,8 +109,8 @@ global: # Configuration for the Appservice API. app_service_api: internal_api: - listen: http://localhost:7777 - connect: http://localhost:7777 + listen: http://localhost:7777 # Only used in polylith deployments + connect: http://localhost:7777 # Only used in polylith deployments database: connection_string: file:appservice.db max_open_conns: 10 @@ -128,8 +128,8 @@ app_service_api: # Configuration for the Client API. client_api: internal_api: - listen: http://localhost:7771 - connect: http://localhost:7771 + listen: http://localhost:7771 # Only used in polylith deployments + connect: http://localhost:7771 # Only used in polylith deployments external_api: listen: http://[::]:8071 @@ -169,14 +169,14 @@ client_api: # Configuration for the EDU server. edu_server: internal_api: - listen: http://localhost:7778 - connect: http://localhost:7778 + listen: http://localhost:7778 # Only used in polylith deployments + connect: http://localhost:7778 # Only used in polylith deployments # Configuration for the Federation API. federation_api: internal_api: - listen: http://localhost:7772 - connect: http://localhost:7772 + listen: http://localhost:7772 # Only used in polylith deployments + connect: http://localhost:7772 # Only used in polylith deployments external_api: listen: http://[::]:8072 @@ -189,8 +189,8 @@ federation_api: # Configuration for the Federation Sender. federation_sender: internal_api: - listen: http://localhost:7775 - connect: http://localhost:7775 + listen: http://localhost:7775 # Only used in polylith deployments + connect: http://localhost:7775 # Only used in polylith deployments database: connection_string: file:federationsender.db max_open_conns: 10 @@ -215,8 +215,8 @@ federation_sender: # Configuration for the Key Server (for end-to-end encryption). key_server: internal_api: - listen: http://localhost:7779 - connect: http://localhost:7779 + listen: http://localhost:7779 # Only used in polylith deployments + connect: http://localhost:7779 # Only used in polylith deployments database: connection_string: file:keyserver.db max_open_conns: 10 @@ -226,8 +226,8 @@ key_server: # Configuration for the Media API. media_api: internal_api: - listen: http://localhost:7774 - connect: http://localhost:7774 + listen: http://localhost:7774 # Only used in polylith deployments + connect: http://localhost:7774 # Only used in polylith deployments external_api: listen: http://[::]:8074 database: @@ -278,8 +278,8 @@ mscs: # Configuration for the Room Server. room_server: internal_api: - listen: http://localhost:7770 - connect: http://localhost:7770 + listen: http://localhost:7770 # Only used in polylith deployments + connect: http://localhost:7770 # Only used in polylith deployments database: connection_string: file:roomserver.db max_open_conns: 10 @@ -289,8 +289,8 @@ room_server: # Configuration for the Signing Key Server (for server signing keys). signing_key_server: internal_api: - listen: http://localhost:7780 - connect: http://localhost:7780 + listen: http://localhost:7780 # Only used in polylith deployments + connect: http://localhost:7780 # Only used in polylith deployments database: connection_string: file:signingkeyserver.db max_open_conns: 10 @@ -316,8 +316,8 @@ signing_key_server: # Configuration for the Sync API. sync_api: internal_api: - listen: http://localhost:7773 - connect: http://localhost:7773 + listen: http://localhost:7773 # Only used in polylith deployments + connect: http://localhost:7773 # Only used in polylith deployments external_api: listen: http://[::]:8073 database: @@ -341,8 +341,8 @@ user_api: # This value can be low if performing tests or on embedded Dendrite instances (e.g WASM builds) # bcrypt_cost: 10 internal_api: - listen: http://localhost:7781 - connect: http://localhost:7781 + listen: http://localhost:7781 # Only used in polylith deployments + connect: http://localhost:7781 # Only used in polylith deployments account_database: connection_string: file:userapi_accounts.db max_open_conns: 10 diff --git a/eduserver/eduserver.go b/eduserver/eduserver.go index 9c3f7ddf6..d0b7697be 100644 --- a/eduserver/eduserver.go +++ b/eduserver/eduserver.go @@ -42,12 +42,12 @@ func NewInternalAPI( ) api.EDUServerInputAPI { cfg := &base.Cfg.EDUServer - _, producer := jetstream.SetupConsumerProducer(&cfg.Matrix.JetStream) + js, _, _ := jetstream.Prepare(&cfg.Matrix.JetStream) return &input.EDUServerInputAPI{ Cache: eduCache, UserAPI: userAPI, - Producer: producer, + JetStream: js, OutputTypingEventTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputTypingEvent), OutputSendToDeviceEventTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputSendToDeviceEvent), OutputReceiptEventTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputReceiptEvent), diff --git a/eduserver/input/input.go b/eduserver/input/input.go index bdc243745..e7501a907 100644 --- a/eduserver/input/input.go +++ b/eduserver/input/input.go @@ -21,12 +21,12 @@ import ( "encoding/json" "time" - "github.com/Shopify/sarama" "github.com/matrix-org/dendrite/eduserver/api" "github.com/matrix-org/dendrite/eduserver/cache" keyapi "github.com/matrix-org/dendrite/keyserver/api" userapi "github.com/matrix-org/dendrite/userapi/api" "github.com/matrix-org/gomatrixserverlib" + "github.com/nats-io/nats.go" "github.com/sirupsen/logrus" ) @@ -43,7 +43,7 @@ type EDUServerInputAPI struct { // The kafka topic to output new key change events to OutputKeyChangeEventTopic string // kafka producer - Producer sarama.SyncProducer + JetStream nats.JetStreamContext // Internal user query API UserAPI userapi.UserInternalAPI // our server name @@ -100,13 +100,11 @@ func (t *EDUServerInputAPI) InputCrossSigningKeyUpdate( "user_id": request.UserID, }).Infof("Producing to topic '%s'", t.OutputKeyChangeEventTopic) - m := &sarama.ProducerMessage{ - Topic: string(t.OutputKeyChangeEventTopic), - Key: sarama.StringEncoder(request.UserID), - Value: sarama.ByteEncoder(eventJSON), - } - - _, _, err = t.Producer.SendMessage(m) + _, err = t.JetStream.PublishMsg(&nats.Msg{ + Subject: t.OutputKeyChangeEventTopic, + Header: nats.Header{}, + Data: eventJSON, + }) return err } @@ -138,13 +136,11 @@ func (t *EDUServerInputAPI) sendTypingEvent(ite *api.InputTypingEvent) error { "typing": ite.Typing, }).Infof("Producing to topic '%s'", t.OutputTypingEventTopic) - m := &sarama.ProducerMessage{ - Topic: string(t.OutputTypingEventTopic), - Key: sarama.StringEncoder(ite.RoomID), - Value: sarama.ByteEncoder(eventJSON), - } - - _, _, err = t.Producer.SendMessage(m) + _, err = t.JetStream.PublishMsg(&nats.Msg{ + Subject: t.OutputTypingEventTopic, + Header: nats.Header{}, + Data: eventJSON, + }) return err } @@ -193,14 +189,10 @@ func (t *EDUServerInputAPI) sendToDeviceEvent(ise *api.InputSendToDeviceEvent) e return err } - m := &sarama.ProducerMessage{ - Topic: string(t.OutputSendToDeviceEventTopic), - Key: sarama.StringEncoder(ote.UserID), - Value: sarama.ByteEncoder(eventJSON), - } - - _, _, err = t.Producer.SendMessage(m) - if err != nil { + if _, err = t.JetStream.PublishMsg(&nats.Msg{ + Subject: t.OutputSendToDeviceEventTopic, + Data: eventJSON, + }); err != nil { logrus.WithError(err).Error("sendToDevice failed t.Producer.SendMessage") return err } @@ -228,11 +220,10 @@ func (t *EDUServerInputAPI) InputReceiptEvent( if err != nil { return err } - m := &sarama.ProducerMessage{ - Topic: t.OutputReceiptEventTopic, - Key: sarama.StringEncoder(request.InputReceiptEvent.RoomID + ":" + request.InputReceiptEvent.UserID), - Value: sarama.ByteEncoder(js), - } - _, _, err = t.Producer.SendMessage(m) + + _, err = t.JetStream.PublishMsg(&nats.Msg{ + Subject: t.OutputReceiptEventTopic, + Data: js, + }) return err } diff --git a/federationapi/routing/send.go b/federationapi/routing/send.go index 4b5f0d660..ba73a5a6a 100644 --- a/federationapi/routing/send.go +++ b/federationapi/routing/send.go @@ -692,6 +692,7 @@ func (t *txnReq) processEvent(ctx context.Context, e *gomatrixserverlib.Event) e }, api.DoNotSendToOtherServers, nil, + true, // asynchronous ) } @@ -734,6 +735,7 @@ withNextEvent: SendAsServer: api.DoNotSendToOtherServers, }, }, + false, ); err != nil { return fmt.Errorf("api.SendEvents: %w", err) } @@ -882,6 +884,7 @@ func (t *txnReq) processEventWithMissingState( resolvedState, backwardsExtremity.Headered(roomVersion), hadEvents, + true, ) if err != nil { return fmt.Errorf("api.SendEventWithState: %w", err) @@ -902,6 +905,7 @@ func (t *txnReq) processEventWithMissingState( append(headeredNewEvents, e.Headered(roomVersion)), api.DoNotSendToOtherServers, nil, + true, ); err != nil { return fmt.Errorf("api.SendEvents: %w", err) } diff --git a/federationapi/routing/threepid.go b/federationapi/routing/threepid.go index 5ba28881c..fb919a591 100644 --- a/federationapi/routing/threepid.go +++ b/federationapi/routing/threepid.go @@ -89,7 +89,7 @@ func CreateInvitesFrom3PIDInvites( } // Send all the events - if err := api.SendEvents(req.Context(), rsAPI, api.KindNew, evs, cfg.Matrix.ServerName, nil); err != nil { + if err := api.SendEvents(req.Context(), rsAPI, api.KindNew, evs, cfg.Matrix.ServerName, nil, false); err != nil { util.GetLogger(req.Context()).WithError(err).Error("SendEvents failed") return jsonerror.InternalServerError() } @@ -180,6 +180,7 @@ func ExchangeThirdPartyInvite( }, cfg.Matrix.ServerName, nil, + false, ); err != nil { util.GetLogger(httpReq.Context()).WithError(err).Error("SendEvents failed") return jsonerror.InternalServerError() diff --git a/federationsender/consumers/eduserver.go b/federationsender/consumers/eduserver.go index 63a367f52..bedc51f2f 100644 --- a/federationsender/consumers/eduserver.go +++ b/federationsender/consumers/eduserver.go @@ -17,115 +17,92 @@ package consumers import ( "context" "encoding/json" - "fmt" - "github.com/Shopify/sarama" "github.com/matrix-org/dendrite/eduserver/api" "github.com/matrix-org/dendrite/federationsender/queue" "github.com/matrix-org/dendrite/federationsender/storage" - "github.com/matrix-org/dendrite/internal" "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/gomatrixserverlib" "github.com/matrix-org/util" + "github.com/nats-io/nats.go" log "github.com/sirupsen/logrus" ) // OutputEDUConsumer consumes events that originate in EDU server. type OutputEDUConsumer struct { - typingConsumer *internal.ContinualConsumer - sendToDeviceConsumer *internal.ContinualConsumer - receiptConsumer *internal.ContinualConsumer - db storage.Database - queues *queue.OutgoingQueues - ServerName gomatrixserverlib.ServerName - TypingTopic string - SendToDeviceTopic string + jetstream nats.JetStreamContext + db storage.Database + queues *queue.OutgoingQueues + ServerName gomatrixserverlib.ServerName + typingTopic string + sendToDeviceTopic string + receiptTopic string } // NewOutputEDUConsumer creates a new OutputEDUConsumer. Call Start() to begin consuming from EDU servers. func NewOutputEDUConsumer( process *process.ProcessContext, cfg *config.FederationSender, - kafkaConsumer sarama.Consumer, + js nats.JetStreamContext, queues *queue.OutgoingQueues, store storage.Database, ) *OutputEDUConsumer { - c := &OutputEDUConsumer{ - typingConsumer: &internal.ContinualConsumer{ - Process: process, - ComponentName: "eduserver/typing", - Topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputTypingEvent), - Consumer: kafkaConsumer, - PartitionStore: store, - }, - sendToDeviceConsumer: &internal.ContinualConsumer{ - Process: process, - ComponentName: "eduserver/sendtodevice", - Topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputSendToDeviceEvent), - Consumer: kafkaConsumer, - PartitionStore: store, - }, - receiptConsumer: &internal.ContinualConsumer{ - Process: process, - ComponentName: "eduserver/receipt", - Topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputReceiptEvent), - Consumer: kafkaConsumer, - PartitionStore: store, - }, + return &OutputEDUConsumer{ + jetstream: js, queues: queues, db: store, ServerName: cfg.Matrix.ServerName, - TypingTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputTypingEvent), - SendToDeviceTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputSendToDeviceEvent), + typingTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputTypingEvent), + sendToDeviceTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputSendToDeviceEvent), + receiptTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputReceiptEvent), } - c.typingConsumer.ProcessMessage = c.onTypingEvent - c.sendToDeviceConsumer.ProcessMessage = c.onSendToDeviceEvent - c.receiptConsumer.ProcessMessage = c.onReceiptEvent - - return c } // Start consuming from EDU servers func (t *OutputEDUConsumer) Start() error { - if err := t.typingConsumer.Start(); err != nil { - return fmt.Errorf("t.typingConsumer.Start: %w", err) + if _, err := t.jetstream.Subscribe(t.typingTopic, t.onTypingEvent); err != nil { + return err } - if err := t.sendToDeviceConsumer.Start(); err != nil { - return fmt.Errorf("t.sendToDeviceConsumer.Start: %w", err) + if _, err := t.jetstream.Subscribe(t.sendToDeviceTopic, t.onSendToDeviceEvent); err != nil { + return err } - if err := t.receiptConsumer.Start(); err != nil { - return fmt.Errorf("t.receiptConsumer.Start: %w", err) + if _, err := t.jetstream.Subscribe(t.receiptTopic, t.onReceiptEvent); err != nil { + return err } return nil } // onSendToDeviceEvent is called in response to a message received on the // send-to-device events topic from the EDU server. -func (t *OutputEDUConsumer) onSendToDeviceEvent(msg *sarama.ConsumerMessage) error { +func (t *OutputEDUConsumer) onSendToDeviceEvent(msg *nats.Msg) { // Extract the send-to-device event from msg. var ote api.OutputSendToDeviceEvent - if err := json.Unmarshal(msg.Value, &ote); err != nil { + if err := json.Unmarshal(msg.Data, &ote); err != nil { log.WithError(err).Errorf("eduserver output log: message parse failed (expected send-to-device)") - return nil + _ = msg.Ack() + return } // only send send-to-device events which originated from us _, originServerName, err := gomatrixserverlib.SplitID('@', ote.Sender) if err != nil { log.WithError(err).WithField("user_id", ote.Sender).Error("Failed to extract domain from send-to-device sender") - return nil + _ = msg.Ack() + return } if originServerName != t.ServerName { log.WithField("other_server", originServerName).Info("Suppressing send-to-device: originated elsewhere") - return nil + _ = msg.Ack() + return } _, destServerName, err := gomatrixserverlib.SplitID('@', ote.UserID) if err != nil { log.WithError(err).WithField("user_id", ote.UserID).Error("Failed to extract domain from send-to-device destination") - return nil + _ = msg.Ack() + return } // Pack the EDU and marshal it @@ -144,38 +121,46 @@ func (t *OutputEDUConsumer) onSendToDeviceEvent(msg *sarama.ConsumerMessage) err }, } if edu.Content, err = json.Marshal(tdm); err != nil { - return err + log.WithError(err).Error("failed to marshal EDU JSON") + _ = msg.Ack() + return } log.Infof("Sending send-to-device message into %q destination queue", destServerName) - return t.queues.SendEDU(edu, t.ServerName, []gomatrixserverlib.ServerName{destServerName}) + if err := t.queues.SendEDU(edu, t.ServerName, []gomatrixserverlib.ServerName{destServerName}); err != nil { + log.WithError(err).Error("failed to send EDU") + } + + _ = msg.Ack() } // onTypingEvent is called in response to a message received on the typing // events topic from the EDU server. -func (t *OutputEDUConsumer) onTypingEvent(msg *sarama.ConsumerMessage) error { +func (t *OutputEDUConsumer) onTypingEvent(msg *nats.Msg) { // Extract the typing event from msg. var ote api.OutputTypingEvent - if err := json.Unmarshal(msg.Value, &ote); err != nil { + if err := json.Unmarshal(msg.Data, &ote); err != nil { // Skip this msg but continue processing messages. log.WithError(err).Errorf("eduserver output log: message parse failed (expected typing)") - return nil + _ = msg.Ack() + return } // only send typing events which originated from us _, typingServerName, err := gomatrixserverlib.SplitID('@', ote.Event.UserID) if err != nil { log.WithError(err).WithField("user_id", ote.Event.UserID).Error("Failed to extract domain from typing sender") - return nil + _ = msg.Ack() + return } if typingServerName != t.ServerName { - log.WithField("other_server", typingServerName).Info("Suppressing typing notif: originated elsewhere") - return nil + return } joined, err := t.db.GetJoinedHosts(context.TODO(), ote.Event.RoomID) if err != nil { - return err + log.WithError(err).WithField("room_id", ote.Event.RoomID).Error("failed to get joined hosts for room") + return } names := make([]gomatrixserverlib.ServerName, len(joined)) @@ -189,36 +174,46 @@ func (t *OutputEDUConsumer) onTypingEvent(msg *sarama.ConsumerMessage) error { "user_id": ote.Event.UserID, "typing": ote.Event.Typing, }); err != nil { - return err + log.WithError(err).Error("failed to marshal EDU JSON") + _ = msg.Ack() + return } - return t.queues.SendEDU(edu, t.ServerName, names) + if err := t.queues.SendEDU(edu, t.ServerName, names); err != nil { + log.WithError(err).Error("failed to send EDU") + } + + _ = msg.Ack() } // onReceiptEvent is called in response to a message received on the receipt // events topic from the EDU server. -func (t *OutputEDUConsumer) onReceiptEvent(msg *sarama.ConsumerMessage) error { +func (t *OutputEDUConsumer) onReceiptEvent(msg *nats.Msg) { // Extract the typing event from msg. var receipt api.OutputReceiptEvent - if err := json.Unmarshal(msg.Value, &receipt); err != nil { + if err := json.Unmarshal(msg.Data, &receipt); err != nil { // Skip this msg but continue processing messages. log.WithError(err).Errorf("eduserver output log: message parse failed (expected receipt)") - return nil + _ = msg.Ack() + return } // only send receipt events which originated from us _, receiptServerName, err := gomatrixserverlib.SplitID('@', receipt.UserID) if err != nil { - log.WithError(err).WithField("user_id", receipt.UserID).Error("Failed to extract domain from receipt sender") - return nil + log.WithError(err).WithField("user_id", receipt.UserID).Error("failed to extract domain from receipt sender") + _ = msg.Ack() + return } if receiptServerName != t.ServerName { - return nil // don't log, very spammy as it logs for each remote receipt + _ = msg.Ack() + return // don't log, very spammy as it logs for each remote receipt } joined, err := t.db.GetJoinedHosts(context.TODO(), receipt.RoomID) if err != nil { - return err + log.WithError(err).WithField("room_id", receipt.RoomID).Error("failed to get joined hosts for room") + return } names := make([]gomatrixserverlib.ServerName, len(joined)) @@ -243,8 +238,14 @@ func (t *OutputEDUConsumer) onReceiptEvent(msg *sarama.ConsumerMessage) error { Origin: string(t.ServerName), } if edu.Content, err = json.Marshal(content); err != nil { - return err + log.WithError(err).Error("failed to marshal EDU JSON") + _ = msg.Ack() + return } - return t.queues.SendEDU(edu, t.ServerName, names) + if err := t.queues.SendEDU(edu, t.ServerName, names); err != nil { + log.WithError(err).Error("failed to send EDU") + } + + _ = msg.Ack() } diff --git a/federationsender/consumers/keychange.go b/federationsender/consumers/keychange.go index 5420bef3f..f86f4c1c6 100644 --- a/federationsender/consumers/keychange.go +++ b/federationsender/consumers/keychange.go @@ -101,6 +101,9 @@ func (t *KeyChangeConsumer) onMessage(msg *sarama.ConsumerMessage) error { } func (t *KeyChangeConsumer) onDeviceKeyMessage(m api.DeviceMessage) error { + if m.DeviceKeys == nil { + return nil + } logger := logrus.WithField("user_id", m.UserID) // only send key change events which originated from us diff --git a/federationsender/consumers/roomserver.go b/federationsender/consumers/roomserver.go index b145bc506..af509d084 100644 --- a/federationsender/consumers/roomserver.go +++ b/federationsender/consumers/roomserver.go @@ -19,72 +19,65 @@ import ( "encoding/json" "fmt" - "github.com/Shopify/sarama" "github.com/matrix-org/dendrite/federationsender/queue" "github.com/matrix-org/dendrite/federationsender/storage" "github.com/matrix-org/dendrite/federationsender/types" - "github.com/matrix-org/dendrite/internal" "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" "github.com/matrix-org/gomatrixserverlib" + "github.com/nats-io/nats.go" log "github.com/sirupsen/logrus" ) // OutputRoomEventConsumer consumes events that originated in the room server. type OutputRoomEventConsumer struct { - cfg *config.FederationSender - rsAPI api.RoomserverInternalAPI - rsConsumer *internal.ContinualConsumer - db storage.Database - queues *queue.OutgoingQueues + cfg *config.FederationSender + rsAPI api.RoomserverInternalAPI + jetstream nats.JetStreamContext + db storage.Database + queues *queue.OutgoingQueues + topic string } // NewOutputRoomEventConsumer creates a new OutputRoomEventConsumer. Call Start() to begin consuming from room servers. func NewOutputRoomEventConsumer( process *process.ProcessContext, cfg *config.FederationSender, - kafkaConsumer sarama.Consumer, + js nats.JetStreamContext, queues *queue.OutgoingQueues, store storage.Database, rsAPI api.RoomserverInternalAPI, ) *OutputRoomEventConsumer { - consumer := internal.ContinualConsumer{ - Process: process, - ComponentName: "federationsender/roomserver", - Topic: string(cfg.Matrix.JetStream.TopicFor(jetstream.OutputRoomEvent)), - Consumer: kafkaConsumer, - PartitionStore: store, + return &OutputRoomEventConsumer{ + cfg: cfg, + jetstream: js, + db: store, + queues: queues, + rsAPI: rsAPI, + topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputRoomEvent), } - s := &OutputRoomEventConsumer{ - cfg: cfg, - rsConsumer: &consumer, - db: store, - queues: queues, - rsAPI: rsAPI, - } - consumer.ProcessMessage = s.onMessage - - return s } // Start consuming from room servers func (s *OutputRoomEventConsumer) Start() error { - return s.rsConsumer.Start() + _, err := s.jetstream.Subscribe(s.topic, s.onMessage) + return err } // onMessage is called when the federation server receives a new event from the room server output log. // It is unsafe to call this with messages for the same room in multiple gorountines // because updates it will likely fail with a types.EventIDMismatchError when it // realises that it cannot update the room state using the deltas. -func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error { +func (s *OutputRoomEventConsumer) onMessage(msg *nats.Msg) { // Parse out the event JSON var output api.OutputEvent - if err := json.Unmarshal(msg.Value, &output); err != nil { + if err := json.Unmarshal(msg.Data, &output); err != nil { // If the message was invalid, log it and move on to the next message in the stream log.WithError(err).Errorf("roomserver output log: message parse failure") - return nil + _ = msg.Ack() + return } switch output.Type { @@ -93,7 +86,8 @@ func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error { if output.NewRoomEvent.RewritesState { if err := s.db.PurgeRoomState(context.TODO(), ev.RoomID()); err != nil { - return fmt.Errorf("s.db.PurgeRoom: %w", err) + log.WithError(err).Errorf("roomserver output log: purge room state failure") + return } } @@ -103,6 +97,7 @@ func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error { log.WithField("error", output.Type).Info( err.Error(), ) + _ = msg.Ack() default: // panic rather than continue with an inconsistent database log.WithFields(log.Fields{ @@ -113,24 +108,28 @@ func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error { log.ErrorKey: err, }).Panicf("roomserver output log: write room event failure") } - return nil + return } + + _ = msg.Ack() + case api.OutputTypeNewInboundPeek: if err := s.processInboundPeek(*output.NewInboundPeek); err != nil { log.WithFields(log.Fields{ "event": output.NewInboundPeek, log.ErrorKey: err, }).Panicf("roomserver output log: remote peek event failure") - return nil + return } + _ = msg.Ack() + default: log.WithField("type", output.Type).Debug( "roomserver output log: ignoring unknown output type", ) - return nil + _ = msg.Ack() + return } - - return nil } // processInboundPeek starts tracking a new federated inbound peek (replacing the existing one if any) diff --git a/federationsender/federationsender.go b/federationsender/federationsender.go index ee89d7822..50b216e5f 100644 --- a/federationsender/federationsender.go +++ b/federationsender/federationsender.go @@ -61,7 +61,7 @@ func NewInternalAPI( FailuresUntilBlacklist: cfg.FederationMaxRetries, } - consumer, _ := jetstream.SetupConsumerProducer(&cfg.Matrix.JetStream) + js, consumer, _ := jetstream.Prepare(&cfg.Matrix.JetStream) queues := queue.NewOutgoingQueues( federationSenderDB, base.ProcessContext, @@ -75,19 +75,19 @@ func NewInternalAPI( ) rsConsumer := consumers.NewOutputRoomEventConsumer( - base.ProcessContext, cfg, consumer, queues, - federationSenderDB, rsAPI, + base.ProcessContext, cfg, js, queues, federationSenderDB, rsAPI, ) if err = rsConsumer.Start(); err != nil { logrus.WithError(err).Panic("failed to start room server consumer") } tsConsumer := consumers.NewOutputEDUConsumer( - base.ProcessContext, cfg, consumer, queues, federationSenderDB, + base.ProcessContext, cfg, js, queues, federationSenderDB, ) if err := tsConsumer.Start(); err != nil { logrus.WithError(err).Panic("failed to start typing server consumer") } + keyConsumer := consumers.NewKeyChangeConsumer( base.ProcessContext, &base.Cfg.KeyServer, consumer, queues, federationSenderDB, rsAPI, ) diff --git a/federationsender/internal/perform.go b/federationsender/internal/perform.go index 53fa974b2..47ec54278 100644 --- a/federationsender/internal/perform.go +++ b/federationsender/internal/perform.go @@ -249,7 +249,7 @@ func (r *FederationSenderInternalAPI) performJoinUsingServer( roomserverAPI.KindNew, respState, event.Headered(respMakeJoin.RoomVersion), - nil, + nil, false, ); err != nil { logrus.WithFields(logrus.Fields{ "room_id": roomID, @@ -430,7 +430,7 @@ func (r *FederationSenderInternalAPI) performOutboundPeekUsingServer( roomserverAPI.KindNew, &respState, respPeek.LatestEvent.Headered(respPeek.RoomVersion), - nil, + nil, false, ); err != nil { return fmt.Errorf("r.producer.SendEventWithState: %w", err) } diff --git a/go.mod b/go.mod index 3991c6b8c..1728ab5d4 100644 --- a/go.mod +++ b/go.mod @@ -6,22 +6,25 @@ replace github.com/nats-io/nats.go => github.com/neilalexander/nats.go v1.11.1-0 require ( github.com/Arceliar/ironwood v0.0.0-20210619124114-6ad55cae5031 + github.com/Arceliar/phony v0.0.0-20210209235338-dde1a8dca979 github.com/DATA-DOG/go-sqlmock v1.5.0 github.com/HdrHistogram/hdrhistogram-go v1.0.1 // indirect github.com/Masterminds/semver/v3 v3.1.1 github.com/S7evinK/saramajetstream v0.0.0-20210709110708-de6efc8c4a32 - github.com/Shopify/sarama v1.29.1 + github.com/Shopify/sarama v1.29.0 github.com/codeclysm/extract v2.2.0+incompatible github.com/containerd/containerd v1.5.7 // indirect github.com/docker/docker v20.10.7+incompatible github.com/docker/go-connections v0.4.0 github.com/getsentry/sentry-go v0.11.0 + github.com/golang/snappy v0.0.4 // indirect github.com/gologme/log v1.2.0 github.com/gorilla/mux v1.8.0 github.com/gorilla/websocket v1.4.2 github.com/h2non/filetype v1.1.1 // indirect github.com/hashicorp/golang-lru v0.5.4 github.com/juju/testing v0.0.0-20210324180055-18c50b0c2098 // indirect + github.com/klauspost/compress v1.13.6 // indirect github.com/lib/pq v1.10.1 github.com/libp2p/go-libp2p v0.13.0 github.com/libp2p/go-libp2p-circuit v0.4.0 @@ -64,6 +67,7 @@ require ( golang.org/x/mobile v0.0.0-20210716004757-34ab1303b554 golang.org/x/net v0.0.0-20210927181540-4e4d966f7476 golang.org/x/term v0.0.0-20210615171337-6886f2dfbf5b + golang.org/x/text v0.3.7 // indirect gopkg.in/h2non/bimg.v1 v1.1.5 gopkg.in/yaml.v2 v2.4.0 nhooyr.io/websocket v1.8.7 diff --git a/go.sum b/go.sum index d455c62c7..422f1acfd 100644 --- a/go.sum +++ b/go.sum @@ -91,9 +91,8 @@ github.com/S7evinK/saramajetstream v0.0.0-20210709110708-de6efc8c4a32 h1:i3fOph9 github.com/S7evinK/saramajetstream v0.0.0-20210709110708-de6efc8c4a32/go.mod h1:ne+jkLlzafIzaE4Q0Ze81T27dNgXe1wxovVEoAtSHTc= github.com/Shopify/goreferrer v0.0.0-20181106222321-ec9c9a553398/go.mod h1:a1uqRtAwp2Xwc6WNPJEufxJ7fx3npB4UV/JOLmbu5I0= github.com/Shopify/logrus-bugsnag v0.0.0-20171204204709-577dee27f20d/go.mod h1:HI8ITrYtUY+O+ZhtlqUnD8+KwNPOyugEhfP9fdUIaEQ= +github.com/Shopify/sarama v1.29.0 h1:ARid8o8oieau9XrHI55f/L3EoRAhm9px6sonbD7yuUE= github.com/Shopify/sarama v1.29.0/go.mod h1:2QpgD79wpdAESqNQMxNc0KYMkycd4slxGdV3TWSVqrU= -github.com/Shopify/sarama v1.29.1 h1:wBAacXbYVLmWieEA/0X/JagDdCZ8NVFOfS6l6+2u5S0= -github.com/Shopify/sarama v1.29.1/go.mod h1:mdtqvCSg8JOxk8PmpTNGyo6wzd4BMm4QXSfDnTXmgkE= github.com/Shopify/toxiproxy v2.1.4+incompatible h1:TKdv8HiTLgE5wdJuEML90aBgNWsokNbMijUGhmcoBJc= github.com/Shopify/toxiproxy v2.1.4+incompatible/go.mod h1:OXgGpZ6Cli1/URJOF1DMxUHB2q5Ap20/P/eIdh4G0pI= github.com/VividCortex/ewma v1.1.1/go.mod h1:2Tkkvm3sRDVXaiyucHiACn4cqf7DpdyLvmxzcbUokwA= @@ -474,8 +473,9 @@ github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaS github.com/golang/protobuf v1.5.2 h1:ROPKBNFfQgOUMifHyP+KYbvpjbdoFNs+aK7DXlji0Tw= github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY= github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= -github.com/golang/snappy v0.0.3 h1:fHPg5GQYlCeLIPB9BZqMVR5nR9A+IM5zcgeTdjMYmLA= github.com/golang/snappy v0.0.3/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= +github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM= +github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= github.com/gologme/log v1.2.0 h1:Ya5Ip/KD6FX7uH0S31QO87nCCSucKtF44TLbTtO7V4c= github.com/gologme/log v1.2.0/go.mod h1:gq31gQ8wEHkR+WekdWsqDuf8pXTUZA9BnnzTuPz1Y9U= github.com/gomodule/redigo v1.7.1-0.20190724094224-574c33c3df38/go.mod h1:B4C85qUVwatsJoIUNIfCRsp7qO0iAmpGFZ4EELWSbC4= @@ -719,8 +719,9 @@ github.com/klauspost/compress v1.11.3/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYs github.com/klauspost/compress v1.11.13/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs= github.com/klauspost/compress v1.12.2/go.mod h1:8dP1Hq4DHOhN9w426knH3Rhby4rFm6D8eO+e+Dq5Gzg= github.com/klauspost/compress v1.12.3/go.mod h1:8dP1Hq4DHOhN9w426knH3Rhby4rFm6D8eO+e+Dq5Gzg= -github.com/klauspost/compress v1.13.4 h1:0zhec2I8zGnjWcKyLl6i3gPqKANCCn5e9xmviEEeX6s= github.com/klauspost/compress v1.13.4/go.mod h1:8dP1Hq4DHOhN9w426knH3Rhby4rFm6D8eO+e+Dq5Gzg= +github.com/klauspost/compress v1.13.6 h1:P76CopJELS0TiO2mebmnzgWaajssP/EszplttgQxcgc= +github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk= github.com/klauspost/cpuid v1.2.1/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgoMS4s3ek= github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= github.com/konsorten/go-windows-terminal-sequences v1.0.2/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= @@ -1602,7 +1603,6 @@ golang.org/x/net v0.0.0-20210428140749-89ef3d95e781/go.mod h1:OJAsFXCWl8Ukc7SiCT golang.org/x/net v0.0.0-20210510120150-4163338589ed/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/net v0.0.0-20210525063256-abc453219eb5/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/net v0.0.0-20210610132358-84b48f89b13b/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= -golang.org/x/net v0.0.0-20210614182718-04defd469f4e/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/net v0.0.0-20210927181540-4e4d966f7476 h1:s5hu7bTnLKswvidgtqc4GwsW83m9LZu8UAqzmWOZtI4= golang.org/x/net v0.0.0-20210927181540-4e4d966f7476/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= @@ -1725,8 +1725,9 @@ golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= -golang.org/x/text v0.3.7-0.20210503195748-5c7c50ebbd4f h1:yQJrRE0hDxDFmZLlRaw+3vusO4fwNHgHIjUOMO7bHYI= golang.org/x/text v0.3.7-0.20210503195748-5c7c50ebbd4f/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= +golang.org/x/text v0.3.7 h1:olpwvP2KacW1ZWvsR7uQhoyTYvKAupfQrRGBFM352Gk= +golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= diff --git a/keyserver/keyserver.go b/keyserver/keyserver.go index dc7f79019..b0d3c764d 100644 --- a/keyserver/keyserver.go +++ b/keyserver/keyserver.go @@ -40,7 +40,7 @@ func AddInternalRoutes(router *mux.Router, intAPI api.KeyInternalAPI) { func NewInternalAPI( base *setup.BaseDendrite, cfg *config.KeyServer, fedClient fedsenderapi.FederationClient, ) api.KeyInternalAPI { - consumer, producer := jetstream.SetupConsumerProducer(&cfg.Matrix.JetStream) + _, consumer, producer := jetstream.Prepare(&cfg.Matrix.JetStream) db, err := storage.NewDatabase(&cfg.Database) if err != nil { diff --git a/roomserver/api/input.go b/roomserver/api/input.go index 8e6e4ac7b..a537e64ef 100644 --- a/roomserver/api/input.go +++ b/roomserver/api/input.go @@ -86,6 +86,7 @@ type TransactionID struct { // InputRoomEventsRequest is a request to InputRoomEvents type InputRoomEventsRequest struct { InputRoomEvents []InputRoomEvent `json:"input_room_events"` + Asynchronous bool `json:"async"` } // InputRoomEventsResponse is a response to InputRoomEvents diff --git a/roomserver/api/wrapper.go b/roomserver/api/wrapper.go index de66df803..cdb186c07 100644 --- a/roomserver/api/wrapper.go +++ b/roomserver/api/wrapper.go @@ -27,6 +27,7 @@ func SendEvents( ctx context.Context, rsAPI RoomserverInternalAPI, kind Kind, events []*gomatrixserverlib.HeaderedEvent, sendAsServer gomatrixserverlib.ServerName, txnID *TransactionID, + async bool, ) error { ires := make([]InputRoomEvent, len(events)) for i, event := range events { @@ -38,7 +39,7 @@ func SendEvents( TransactionID: txnID, } } - return SendInputRoomEvents(ctx, rsAPI, ires) + return SendInputRoomEvents(ctx, rsAPI, ires, async) } // SendEventWithState writes an event with the specified kind to the roomserver @@ -47,7 +48,7 @@ func SendEvents( func SendEventWithState( ctx context.Context, rsAPI RoomserverInternalAPI, kind Kind, state *gomatrixserverlib.RespState, event *gomatrixserverlib.HeaderedEvent, - haveEventIDs map[string]bool, + haveEventIDs map[string]bool, async bool, ) error { outliers, err := state.Events() if err != nil { @@ -79,14 +80,18 @@ func SendEventWithState( StateEventIDs: stateEventIDs, }) - return SendInputRoomEvents(ctx, rsAPI, ires) + return SendInputRoomEvents(ctx, rsAPI, ires, async) } // SendInputRoomEvents to the roomserver. func SendInputRoomEvents( - ctx context.Context, rsAPI RoomserverInternalAPI, ires []InputRoomEvent, + ctx context.Context, rsAPI RoomserverInternalAPI, + ires []InputRoomEvent, async bool, ) error { - request := InputRoomEventsRequest{InputRoomEvents: ires} + request := InputRoomEventsRequest{ + InputRoomEvents: ires, + Asynchronous: async, + } var response InputRoomEventsResponse rsAPI.InputRoomEvents(ctx, &request, &response) return response.Err() diff --git a/roomserver/internal/api.go b/roomserver/internal/api.go index f39b26eaf..1f66722f7 100644 --- a/roomserver/internal/api.go +++ b/roomserver/internal/api.go @@ -3,7 +3,6 @@ package internal import ( "context" - "github.com/Shopify/sarama" "github.com/getsentry/sentry-go" asAPI "github.com/matrix-org/dendrite/appservice/api" fsAPI "github.com/matrix-org/dendrite/federationsender/api" @@ -16,6 +15,8 @@ import ( "github.com/matrix-org/dendrite/roomserver/storage" "github.com/matrix-org/dendrite/setup/config" "github.com/matrix-org/gomatrixserverlib" + "github.com/nats-io/nats.go" + "github.com/sirupsen/logrus" ) // RoomserverInternalAPI is an implementation of api.RoomserverInternalAPI @@ -33,19 +34,19 @@ type RoomserverInternalAPI struct { *perform.Forgetter DB storage.Database Cfg *config.RoomServer - Producer sarama.SyncProducer Cache caching.RoomServerCaches ServerName gomatrixserverlib.ServerName KeyRing gomatrixserverlib.JSONVerifier fsAPI fsAPI.FederationSenderInternalAPI asAPI asAPI.AppServiceQueryAPI - OutputRoomEventTopic string // Kafka topic for new output room events + InputRoomEventTopic string // JetStream topic for new input room events + OutputRoomEventTopic string // JetStream topic for new output room events PerspectiveServerNames []gomatrixserverlib.ServerName } func NewRoomserverAPI( - cfg *config.RoomServer, roomserverDB storage.Database, producer sarama.SyncProducer, - outputRoomEventTopic string, caches caching.RoomServerCaches, + cfg *config.RoomServer, roomserverDB storage.Database, consumer nats.JetStreamContext, + inputRoomEventTopic, outputRoomEventTopic string, caches caching.RoomServerCaches, keyRing gomatrixserverlib.JSONVerifier, perspectiveServerNames []gomatrixserverlib.ServerName, ) *RoomserverInternalAPI { serverACLs := acls.NewServerACLs(roomserverDB) @@ -64,13 +65,17 @@ func NewRoomserverAPI( }, Inputer: &input.Inputer{ DB: roomserverDB, + InputRoomEventTopic: inputRoomEventTopic, OutputRoomEventTopic: outputRoomEventTopic, - Producer: producer, + JetStream: consumer, ServerName: cfg.Matrix.ServerName, ACLs: serverACLs, }, // perform-er structs get initialised when we have a federation sender to use } + if err := a.Inputer.Start(); err != nil { + logrus.WithError(err).Panic("failed to start roomserver input API") + } return a } diff --git a/roomserver/internal/input/input.go b/roomserver/internal/input/input.go index de40e133d..f561e8b59 100644 --- a/roomserver/internal/input/input.go +++ b/roomserver/internal/input/input.go @@ -19,19 +19,19 @@ import ( "context" "encoding/json" "sync" - "time" - "github.com/Shopify/sarama" + "github.com/Arceliar/phony" "github.com/getsentry/sentry-go" "github.com/matrix-org/dendrite/internal/hooks" "github.com/matrix-org/dendrite/roomserver/acls" "github.com/matrix-org/dendrite/roomserver/api" "github.com/matrix-org/dendrite/roomserver/storage" + "github.com/matrix-org/dendrite/setup/jetstream" "github.com/matrix-org/gomatrixserverlib" + "github.com/nats-io/nats.go" "github.com/prometheus/client_golang/prometheus" log "github.com/sirupsen/logrus" "github.com/tidwall/gjson" - "go.uber.org/atomic" ) var keyContentFields = map[string]string{ @@ -42,105 +42,150 @@ var keyContentFields = map[string]string{ type Inputer struct { DB storage.Database - Producer sarama.SyncProducer + JetStream nats.JetStreamContext ServerName gomatrixserverlib.ServerName ACLs *acls.ServerACLs + InputRoomEventTopic string OutputRoomEventTopic string - workers sync.Map // room ID -> *inputWorker + workers sync.Map // room ID -> *phony.Inbox } -type inputTask struct { - ctx context.Context - event *api.InputRoomEvent - wg *sync.WaitGroup - err error // written back by worker, only safe to read when all tasks are done -} - -type inputWorker struct { - r *Inputer - running atomic.Bool - input *fifoQueue -} - -// Guarded by a CAS on w.running -func (w *inputWorker) start() { - defer w.running.Store(false) - for { - select { - case <-w.input.wait(): - task, ok := w.input.pop() - if !ok { - continue +// onMessage is called when a new event arrives in the roomserver input stream. +func (r *Inputer) Start() error { + _, err := r.JetStream.Subscribe( + r.InputRoomEventTopic, + func(msg *nats.Msg) { + _ = msg.InProgress() + roomID := msg.Header.Get("room_id") + defer roomserverInputBackpressure.With(prometheus.Labels{"room_id": roomID}).Dec() + var inputRoomEvent api.InputRoomEvent + if err := json.Unmarshal(msg.Data, &inputRoomEvent); err != nil { + _ = msg.Term() + return } - roomserverInputBackpressure.With(prometheus.Labels{ - "room_id": task.event.Event.RoomID(), - }).Dec() - hooks.Run(hooks.KindNewEventReceived, task.event.Event) - _, task.err = w.r.processRoomEvent(task.ctx, task.event) - if task.err == nil { - hooks.Run(hooks.KindNewEventPersisted, task.event.Event) - } else { - sentry.CaptureException(task.err) + inbox, _ := r.workers.LoadOrStore(roomID, &phony.Inbox{}) + inbox.(*phony.Inbox).Act(nil, func() { + _ = msg.InProgress() + if _, err := r.processRoomEvent(context.TODO(), &inputRoomEvent); err != nil { + sentry.CaptureException(err) + _ = msg.Respond([]byte(err.Error())) + } else { + hooks.Run(hooks.KindNewEventPersisted, inputRoomEvent.Event) + } + _ = msg.Ack() + }) + }, + nats.ManualAck(), + nats.MaxDeliver(0), + ) + return err +} + +// InputRoomEvents implements api.RoomserverInternalAPI +func (r *Inputer) InputRoomEvents( + ctx context.Context, + request *api.InputRoomEventsRequest, + response *api.InputRoomEventsResponse, +) { + if request.Asynchronous { + var err error + for _, e := range request.InputRoomEvents { + msg := &nats.Msg{ + Subject: r.InputRoomEventTopic, + Header: nats.Header{}, + } + roomID := e.Event.RoomID() + msg.Header.Set("room_id", roomID) + msg.Data, err = json.Marshal(e) + if err != nil { + response.ErrMsg = err.Error() + return + } + if _, err = r.JetStream.PublishMsg(msg); err != nil { + return + } + roomserverInputBackpressure.With(prometheus.Labels{"room_id": roomID}).Inc() + } + } else { + responses := make(chan error, len(request.InputRoomEvents)) + defer close(responses) + for _, e := range request.InputRoomEvents { + inputRoomEvent := e + inbox, _ := r.workers.LoadOrStore(inputRoomEvent.Event.RoomID(), &phony.Inbox{}) + inbox.(*phony.Inbox).Act(nil, func() { + _, err := r.processRoomEvent(context.TODO(), &inputRoomEvent) + if err != nil { + sentry.CaptureException(err) + } else { + hooks.Run(hooks.KindNewEventPersisted, inputRoomEvent.Event) + } + responses <- err + }) + } + for i := 0; i < len(request.InputRoomEvents); i++ { + select { + case <-ctx.Done(): + return + case err := <-responses: + if err != nil { + response.ErrMsg = err.Error() + return + } } - task.wg.Done() - case <-time.After(time.Second * 5): - return } } } // WriteOutputEvents implements OutputRoomEventWriter func (r *Inputer) WriteOutputEvents(roomID string, updates []api.OutputEvent) error { - messages := make([]*sarama.ProducerMessage, len(updates)) - for i := range updates { - value, err := json.Marshal(updates[i]) + var err error + for _, update := range updates { + msg := &nats.Msg{ + Subject: r.OutputRoomEventTopic, + Header: nats.Header{}, + } + msg.Header.Set(jetstream.RoomID, roomID) + msg.Data, err = json.Marshal(update) if err != nil { return err } logger := log.WithFields(log.Fields{ "room_id": roomID, - "type": updates[i].Type, + "type": update.Type, }) - if updates[i].NewRoomEvent != nil { - eventType := updates[i].NewRoomEvent.Event.Type() + if update.NewRoomEvent != nil { + eventType := update.NewRoomEvent.Event.Type() logger = logger.WithFields(log.Fields{ "event_type": eventType, - "event_id": updates[i].NewRoomEvent.Event.EventID(), - "adds_state": len(updates[i].NewRoomEvent.AddsStateEventIDs), - "removes_state": len(updates[i].NewRoomEvent.RemovesStateEventIDs), - "send_as_server": updates[i].NewRoomEvent.SendAsServer, - "sender": updates[i].NewRoomEvent.Event.Sender(), + "event_id": update.NewRoomEvent.Event.EventID(), + "adds_state": len(update.NewRoomEvent.AddsStateEventIDs), + "removes_state": len(update.NewRoomEvent.RemovesStateEventIDs), + "send_as_server": update.NewRoomEvent.SendAsServer, + "sender": update.NewRoomEvent.Event.Sender(), }) - if updates[i].NewRoomEvent.Event.StateKey() != nil { - logger = logger.WithField("state_key", *updates[i].NewRoomEvent.Event.StateKey()) + if update.NewRoomEvent.Event.StateKey() != nil { + logger = logger.WithField("state_key", *update.NewRoomEvent.Event.StateKey()) } contentKey := keyContentFields[eventType] if contentKey != "" { - value := gjson.GetBytes(updates[i].NewRoomEvent.Event.Content(), contentKey) + value := gjson.GetBytes(update.NewRoomEvent.Event.Content(), contentKey) if value.Exists() { logger = logger.WithField("content_value", value.String()) } } - if eventType == "m.room.server_acl" && updates[i].NewRoomEvent.Event.StateKeyEquals("") { - ev := updates[i].NewRoomEvent.Event.Unwrap() + if eventType == "m.room.server_acl" && update.NewRoomEvent.Event.StateKeyEquals("") { + ev := update.NewRoomEvent.Event.Unwrap() defer r.ACLs.OnServerACLUpdate(ev) } } - logger.Infof("Producing to topic '%s'", r.OutputRoomEventTopic) - messages[i] = &sarama.ProducerMessage{ - Topic: r.OutputRoomEventTopic, - Key: sarama.StringEncoder(roomID), - Value: sarama.ByteEncoder(value), + logger.Tracef("Producing to topic '%s'", r.OutputRoomEventTopic) + if _, err := r.JetStream.PublishMsg(msg); err != nil { + logger.WithError(err).Errorf("Failed to produce to topic '%s': %s", r.OutputRoomEventTopic, err) + return err } } - errs := r.Producer.SendMessages(messages) - if errs != nil { - for _, err := range errs.(sarama.ProducerErrors) { - log.WithError(err).WithField("message_bytes", err.Msg.Value.Length()).Error("Write to kafka failed") - } - } - return errs + return nil } func init() { @@ -156,67 +201,3 @@ var roomserverInputBackpressure = prometheus.NewGaugeVec( }, []string{"room_id"}, ) - -// InputRoomEvents implements api.RoomserverInternalAPI -func (r *Inputer) InputRoomEvents( - _ context.Context, - request *api.InputRoomEventsRequest, - response *api.InputRoomEventsResponse, -) { - // Create a wait group. Each task that we dispatch will call Done on - // this wait group so that we know when all of our events have been - // processed. - wg := &sync.WaitGroup{} - wg.Add(len(request.InputRoomEvents)) - tasks := make([]*inputTask, len(request.InputRoomEvents)) - - for i, e := range request.InputRoomEvents { - // Work out if we are running per-room workers or if we're just doing - // it on a global basis (e.g. SQLite). - roomID := "global" - if r.DB.SupportsConcurrentRoomInputs() { - roomID = e.Event.RoomID() - } - - // Look up the worker, or create it if it doesn't exist. This channel - // is buffered to reduce the chance that we'll be blocked by another - // room - the channel will be quite small as it's just pointer types. - w, _ := r.workers.LoadOrStore(roomID, &inputWorker{ - r: r, - input: newFIFOQueue(), - }) - worker := w.(*inputWorker) - - // Create a task. This contains the input event and a reference to - // the wait group, so that the worker can notify us when this specific - // task has been finished. - tasks[i] = &inputTask{ - ctx: context.Background(), - event: &request.InputRoomEvents[i], - wg: wg, - } - - // Send the task to the worker. - if worker.running.CAS(false, true) { - go worker.start() - } - worker.input.push(tasks[i]) - roomserverInputBackpressure.With(prometheus.Labels{ - "room_id": roomID, - }).Inc() - } - - // Wait for all of the workers to return results about our tasks. - wg.Wait() - - // If any of the tasks returned an error, we should probably report - // that back to the caller. - for _, task := range tasks { - if task.err != nil { - response.ErrMsg = task.err.Error() - _, rejected := task.err.(*gomatrixserverlib.NotAllowed) - response.NotAllowed = rejected - return - } - } -} diff --git a/roomserver/internal/input/input_fifo.go b/roomserver/internal/input/input_fifo.go deleted file mode 100644 index 694b17245..000000000 --- a/roomserver/internal/input/input_fifo.go +++ /dev/null @@ -1,64 +0,0 @@ -package input - -import ( - "sync" -) - -type fifoQueue struct { - tasks []*inputTask - count int - mutex sync.Mutex - notifs chan struct{} -} - -func newFIFOQueue() *fifoQueue { - q := &fifoQueue{ - notifs: make(chan struct{}, 1), - } - return q -} - -func (q *fifoQueue) push(frame *inputTask) { - q.mutex.Lock() - defer q.mutex.Unlock() - q.tasks = append(q.tasks, frame) - q.count++ - select { - case q.notifs <- struct{}{}: - default: - } -} - -// pop returns the first item of the queue, if there is one. -// The second return value will indicate if a task was returned. -// You must check this value, even after calling wait(). -func (q *fifoQueue) pop() (*inputTask, bool) { - q.mutex.Lock() - defer q.mutex.Unlock() - if q.count == 0 { - return nil, false - } - frame := q.tasks[0] - q.tasks[0] = nil - q.tasks = q.tasks[1:] - q.count-- - if q.count == 0 { - // Force a GC of the underlying array, since it might have - // grown significantly if the queue was hammered for some reason - q.tasks = nil - } - return frame, true -} - -// wait returns a channel which can be used to detect when an -// item is waiting in the queue. -func (q *fifoQueue) wait() <-chan struct{} { - q.mutex.Lock() - defer q.mutex.Unlock() - if q.count > 0 && len(q.notifs) == 0 { - ch := make(chan struct{}) - close(ch) - return ch - } - return q.notifs -} diff --git a/roomserver/roomserver.go b/roomserver/roomserver.go index 192a056b8..bea5d7f6f 100644 --- a/roomserver/roomserver.go +++ b/roomserver/roomserver.go @@ -41,8 +41,6 @@ func NewInternalAPI( ) api.RoomserverInternalAPI { cfg := &base.Cfg.RoomServer - _, producer := jetstream.SetupConsumerProducer(&cfg.Matrix.JetStream) - var perspectiveServerNames []gomatrixserverlib.ServerName for _, kp := range base.Cfg.SigningKeyServer.KeyPerspectives { perspectiveServerNames = append(perspectiveServerNames, kp.ServerName) @@ -53,8 +51,12 @@ func NewInternalAPI( logrus.WithError(err).Panicf("failed to connect to room server db") } + js, _, _ := jetstream.Prepare(&cfg.Matrix.JetStream) + return internal.NewRoomserverAPI( - cfg, roomserverDB, producer, string(cfg.Matrix.JetStream.TopicFor(jetstream.OutputRoomEvent)), + cfg, roomserverDB, js, + cfg.Matrix.JetStream.TopicFor(jetstream.InputRoomEvent), + cfg.Matrix.JetStream.TopicFor(jetstream.OutputRoomEvent), base.Caches, keyRing, perspectiveServerNames, ) } diff --git a/roomserver/roomserver_test.go b/roomserver/roomserver_test.go deleted file mode 100644 index 335d2a4ca..000000000 --- a/roomserver/roomserver_test.go +++ /dev/null @@ -1,408 +0,0 @@ -package roomserver - -import ( - "bytes" - "context" - "crypto/ed25519" - "encoding/json" - "fmt" - "os" - "reflect" - "testing" - "time" - - "github.com/Shopify/sarama" - "github.com/matrix-org/dendrite/internal/caching" - "github.com/matrix-org/dendrite/internal/test" - "github.com/matrix-org/dendrite/roomserver/api" - "github.com/matrix-org/dendrite/roomserver/internal" - "github.com/matrix-org/dendrite/roomserver/storage" - "github.com/matrix-org/dendrite/setup" - "github.com/matrix-org/dendrite/setup/config" - "github.com/matrix-org/dendrite/setup/jetstream" - "github.com/matrix-org/gomatrixserverlib" - "github.com/sirupsen/logrus" -) - -const ( - testOrigin = gomatrixserverlib.ServerName("kaer.morhen") - // we have to use an on-disk DB because we open multiple connections due to the *Updater structs. - // Using :memory: results in a brand new DB for each open connection, and sharing memory via - // ?cache=shared just allows read-only sharing, so writes to the database on other connections are lost. - roomserverDBFileURI = "file:roomserver_test.db" - roomserverDBFilePath = "./roomserver_test.db" -) - -var ( - ctx = context.Background() -) - -type dummyProducer struct { - topic string - producedMessages []*api.OutputEvent -} - -// SendMessage produces a given message, and returns only when it either has -// succeeded or failed to produce. It will return the partition and the offset -// of the produced message, or an error if the message failed to produce. -func (p *dummyProducer) SendMessage(msg *sarama.ProducerMessage) (partition int32, offset int64, err error) { - if msg.Topic != p.topic { - return 0, 0, nil - } - be := msg.Value.(sarama.ByteEncoder) - b := json.RawMessage(be) - fmt.Println("SENDING >>>>>>>> ", string(b)) - var out api.OutputEvent - err = json.Unmarshal(b, &out) - if err != nil { - return 0, 0, err - } - p.producedMessages = append(p.producedMessages, &out) - return 0, 0, nil -} - -// SendMessages produces a given set of messages, and returns only when all -// messages in the set have either succeeded or failed. Note that messages -// can succeed and fail individually; if some succeed and some fail, -// SendMessages will return an error. -func (p *dummyProducer) SendMessages(msgs []*sarama.ProducerMessage) error { - for _, m := range msgs { - p.SendMessage(m) - } - return nil -} - -// Close shuts down the producer and waits for any buffered messages to be -// flushed. You must call this function before a producer object passes out of -// scope, as it may otherwise leak memory. You must call this before calling -// Close on the underlying client. -func (p *dummyProducer) Close() error { - return nil -} - -func deleteDatabase() { - err := os.Remove(roomserverDBFilePath) - if err != nil { - fmt.Printf("failed to delete database %s: %s\n", roomserverDBFilePath, err) - } -} - -type fledglingEvent struct { - Type string - StateKey *string - Content interface{} - Sender string - RoomID string -} - -func mustCreateEvents(t *testing.T, roomVer gomatrixserverlib.RoomVersion, events []fledglingEvent) (result []*gomatrixserverlib.HeaderedEvent) { - t.Helper() - depth := int64(1) - seed := make([]byte, ed25519.SeedSize) // zero seed - key := ed25519.NewKeyFromSeed(seed) - var prevs []string - roomState := make(map[gomatrixserverlib.StateKeyTuple]string) // state -> event ID - for _, ev := range events { - eb := gomatrixserverlib.EventBuilder{ - Sender: ev.Sender, - Depth: depth, - Type: ev.Type, - StateKey: ev.StateKey, - RoomID: ev.RoomID, - PrevEvents: prevs, - } - err := eb.SetContent(ev.Content) - if err != nil { - t.Fatalf("mustCreateEvent: failed to marshal event content %+v", ev.Content) - } - stateNeeded, err := gomatrixserverlib.StateNeededForEventBuilder(&eb) - if err != nil { - t.Fatalf("mustCreateEvent: failed to work out auth_events : %s", err) - } - var authEvents []string - for _, tuple := range stateNeeded.Tuples() { - eventID := roomState[tuple] - if eventID != "" { - authEvents = append(authEvents, eventID) - } - } - eb.AuthEvents = authEvents - signedEvent, err := eb.Build(time.Now(), testOrigin, "ed25519:test", key, roomVer) - if err != nil { - t.Fatalf("mustCreateEvent: failed to sign event: %s", err) - } - depth++ - prevs = []string{signedEvent.EventID()} - if ev.StateKey != nil { - roomState[gomatrixserverlib.StateKeyTuple{ - EventType: ev.Type, - StateKey: *ev.StateKey, - }] = signedEvent.EventID() - } - result = append(result, signedEvent.Headered(roomVer)) - } - return -} - -func mustLoadRawEvents(t *testing.T, ver gomatrixserverlib.RoomVersion, events []json.RawMessage) []*gomatrixserverlib.HeaderedEvent { - t.Helper() - hs := make([]*gomatrixserverlib.HeaderedEvent, len(events)) - for i := range events { - e, err := gomatrixserverlib.NewEventFromTrustedJSON(events[i], false, ver) - if err != nil { - t.Fatalf("cannot load test data: " + err.Error()) - } - hs[i] = e.Headered(ver) - } - return hs -} - -func mustCreateRoomserverAPI(t *testing.T) (api.RoomserverInternalAPI, *dummyProducer) { - t.Helper() - cfg := &config.Dendrite{} - cfg.Defaults() - cfg.Global.ServerName = testOrigin - cfg.RoomServer.Database = config.DatabaseOptions{ - ConnectionString: roomserverDBFileURI, - } - dp := &dummyProducer{ - topic: cfg.Global.JetStream.TopicFor(jetstream.OutputRoomEvent), - } - cache, err := caching.NewInMemoryLRUCache(false) - if err != nil { - t.Fatalf("failed to make caches: %s", err) - } - base := &setup.BaseDendrite{ - Caches: cache, - Cfg: cfg, - } - roomserverDB, err := storage.Open(&cfg.RoomServer.Database, base.Caches) - if err != nil { - logrus.WithError(err).Panicf("failed to connect to room server db") - } - return internal.NewRoomserverAPI( - &cfg.RoomServer, roomserverDB, dp, string(cfg.Global.JetStream.TopicFor(jetstream.OutputRoomEvent)), - base.Caches, &test.NopJSONVerifier{}, nil, - ), dp -} - -func mustSendEvents(t *testing.T, ver gomatrixserverlib.RoomVersion, events []json.RawMessage) (api.RoomserverInternalAPI, *dummyProducer, []*gomatrixserverlib.HeaderedEvent) { - t.Helper() - rsAPI, dp := mustCreateRoomserverAPI(t) - hevents := mustLoadRawEvents(t, ver, events) - if err := api.SendEvents(ctx, rsAPI, api.KindNew, hevents, testOrigin, nil); err != nil { - t.Errorf("failed to SendEvents: %s", err) - } - return rsAPI, dp, hevents -} - -func TestOutputRedactedEvent(t *testing.T) { - redactionEvents := []json.RawMessage{ - // create event - []byte(`{"auth_events":[],"content":{"creator":"@userid:kaer.morhen"},"depth":0,"event_id":"$N4us6vqqq3RjvpKd:kaer.morhen","hashes":{"sha256":"WTdrCn/YsiounXcJPsLP8xT0ZjHiO5Ov0NvXYmK2onE"},"origin":"kaer.morhen","origin_server_ts":0,"prev_events":[],"prev_state":[],"room_id":"!roomid:kaer.morhen","sender":"@userid:kaer.morhen","signatures":{"kaer.morhen":{"ed25519:auto":"9+5JcpaN5b5KlHYHGp6r+GoNDH98lbfzGYwjfxensa5C5D/bDACaYnMDLnhwsHOE5nxgI+jT/GV271pz6PMSBQ"}},"state_key":"","type":"m.room.create"}`), - // join event - []byte(`{"auth_events":[["$N4us6vqqq3RjvpKd:kaer.morhen",{"sha256":"SylirfgfXFhscZL7p10NmOa1nFFEckiwz0lAideQMIM"}]],"content":{"membership":"join"},"depth":1,"event_id":"$6sUiGPQ0a3tqYGKo:kaer.morhen","hashes":{"sha256":"eYVBC7RO+FlxRyW1aXYf/ad4Dzi7T93tArdGw3r4RwQ"},"origin":"kaer.morhen","origin_server_ts":0,"prev_events":[["$N4us6vqqq3RjvpKd:kaer.morhen",{"sha256":"SylirfgfXFhscZL7p10NmOa1nFFEckiwz0lAideQMIM"}]],"prev_state":[],"room_id":"!roomid:kaer.morhen","sender":"@userid:kaer.morhen","signatures":{"kaer.morhen":{"ed25519:auto":"tiDBTPFa53YMfHiupX3vSRE/ZcCiCjmGt7gDpIpDpwZapeays5Vqqcqb7KiywrDldpTkrrdJBAw2jXcq6ZyhDw"}},"state_key":"@userid:kaer.morhen","type":"m.room.member"}`), - // room name - []byte(`{"auth_events":[["$N4us6vqqq3RjvpKd:kaer.morhen",{"sha256":"SylirfgfXFhscZL7p10NmOa1nFFEckiwz0lAideQMIM"}],["$6sUiGPQ0a3tqYGKo:kaer.morhen",{"sha256":"IS4HSMqpqVUGh1Z3qgC99YcaizjCoO4yFhYYe8j53IE"}]],"content":{"name":"My Room Name"},"depth":2,"event_id":"$VC1zZ9YWwuUbSNHD:kaer.morhen","hashes":{"sha256":"bpqTkfLx6KHzWz7/wwpsXnXwJWEGW14aV63ffexzDFg"},"origin":"kaer.morhen","origin_server_ts":0,"prev_events":[["$6sUiGPQ0a3tqYGKo:kaer.morhen",{"sha256":"IS4HSMqpqVUGh1Z3qgC99YcaizjCoO4yFhYYe8j53IE"}]],"prev_state":[],"room_id":"!roomid:kaer.morhen","sender":"@userid:kaer.morhen","signatures":{"kaer.morhen":{"ed25519:auto":"mhJZ3X4bAKrF/T0mtPf1K2Tmls0h6xGY1IPDpJ/SScQBqDlu3HQR2BPa7emqj5bViyLTWVNh+ZCpzx/6STTrAg"}},"state_key":"","type":"m.room.name"}`), - // redact room name - []byte(`{"auth_events":[["$N4us6vqqq3RjvpKd:kaer.morhen",{"sha256":"SylirfgfXFhscZL7p10NmOa1nFFEckiwz0lAideQMIM"}],["$6sUiGPQ0a3tqYGKo:kaer.morhen",{"sha256":"IS4HSMqpqVUGh1Z3qgC99YcaizjCoO4yFhYYe8j53IE"}]],"content":{"reason":"Spamming"},"depth":3,"event_id":"$tJI0pE3b8u9UMYpT:kaer.morhen","hashes":{"sha256":"/3TStqa5SQqYaEtl7ajEvSRvu6d12MMKfICUzrBpd2Q"},"origin":"kaer.morhen","origin_server_ts":0,"prev_events":[["$VC1zZ9YWwuUbSNHD:kaer.morhen",{"sha256":"+l8cNa7syvm0EF7CAmQRlYknLEMjivnI4FLhB/TUBEY"}]],"redacts":"$VC1zZ9YWwuUbSNHD:kaer.morhen","room_id":"!roomid:kaer.morhen","sender":"@userid:kaer.morhen","signatures":{"kaer.morhen":{"ed25519:auto":"QBOh+amf0vTJbm6+9VwAcR9uJviBIor2KON0Y7+EyQx5YbUZEzW1HPeJxarLIHBcxMzgOVzjuM+StzjbUgDzAg"}},"type":"m.room.redaction"}`), - // message - []byte(`{"auth_events":[["$N4us6vqqq3RjvpKd:kaer.morhen",{"sha256":"SylirfgfXFhscZL7p10NmOa1nFFEckiwz0lAideQMIM"}],["$6sUiGPQ0a3tqYGKo:kaer.morhen",{"sha256":"IS4HSMqpqVUGh1Z3qgC99YcaizjCoO4yFhYYe8j53IE"}]],"content":{"body":"Test Message"},"depth":4,"event_id":"$o8KHsgSIYbJrddnd:kaer.morhen","hashes":{"sha256":"IE/rGVlKOpiGWeIo887g1CK1drYqcWDZhL6THZHkJ1c"},"origin":"kaer.morhen","origin_server_ts":0,"prev_events":[["$tJI0pE3b8u9UMYpT:kaer.morhen",{"sha256":"zvmwyXuDox7jpA16JRH6Fc1zbfQht2tpkBbMTUOi3Jw"}]],"room_id":"!roomid:kaer.morhen","sender":"@userid:kaer.morhen","signatures":{"kaer.morhen":{"ed25519:auto":"/3z+pJjiJXWhwfqIEzmNksvBHCoXTktK/y0rRuWJXw6i1+ygRG/suDCKhFuuz6gPapRmEMPVILi2mJqHHXPKAg"}},"type":"m.room.message"}`), - // redact previous message - []byte(`{"auth_events":[["$N4us6vqqq3RjvpKd:kaer.morhen",{"sha256":"SylirfgfXFhscZL7p10NmOa1nFFEckiwz0lAideQMIM"}],["$6sUiGPQ0a3tqYGKo:kaer.morhen",{"sha256":"IS4HSMqpqVUGh1Z3qgC99YcaizjCoO4yFhYYe8j53IE"}]],"content":{"reason":"Spamming more"},"depth":5,"event_id":"$UpsE8belb2gJItJG:kaer.morhen","hashes":{"sha256":"zU8PWJOld/I7OtjdpltFSKC+DMNm2ZyEXAHcprsafD0"},"origin":"kaer.morhen","origin_server_ts":0,"prev_events":[["$o8KHsgSIYbJrddnd:kaer.morhen",{"sha256":"UgjMuCFXH4warIjKuwlRq9zZ6dSJrZWCd+CkqtgLSHM"}]],"redacts":"$o8KHsgSIYbJrddnd:kaer.morhen","room_id":"!roomid:kaer.morhen","sender":"@userid:kaer.morhen","signatures":{"kaer.morhen":{"ed25519:auto":"zxFGr/7aGOzqOEN6zRNrBpFkkMnfGFPbCteYL33wC+PycBPIK+2WRa5qlAR2+lcLiK3HjIzwRYkKNsVFTqvRAw"}},"type":"m.room.redaction"}`), - } - var redactedOutputs []api.OutputEvent - deleteDatabase() - _, producer, hevents := mustSendEvents(t, gomatrixserverlib.RoomVersionV1, redactionEvents) - defer deleteDatabase() - for _, msg := range producer.producedMessages { - if msg.Type == api.OutputTypeRedactedEvent { - redactedOutputs = append(redactedOutputs, *msg) - } - } - wantRedactedOutputs := []api.OutputEvent{ - { - Type: api.OutputTypeRedactedEvent, - RedactedEvent: &api.OutputRedactedEvent{ - RedactedEventID: hevents[2].EventID(), - RedactedBecause: hevents[3], - }, - }, - { - Type: api.OutputTypeRedactedEvent, - RedactedEvent: &api.OutputRedactedEvent{ - RedactedEventID: hevents[4].EventID(), - RedactedBecause: hevents[5], - }, - }, - } - t.Logf("redactedOutputs: %+v", redactedOutputs) - if len(wantRedactedOutputs) != len(redactedOutputs) { - t.Fatalf("Got %d redacted events, want %d", len(redactedOutputs), len(wantRedactedOutputs)) - } - for i := 0; i < len(wantRedactedOutputs); i++ { - if !reflect.DeepEqual(*redactedOutputs[i].RedactedEvent, *wantRedactedOutputs[i].RedactedEvent) { - t.Errorf("OutputRedactionEvent %d: wrong event got:\n%+v want:\n%+v", i+1, redactedOutputs[i].RedactedEvent, wantRedactedOutputs[i].RedactedEvent) - } - } -} - -// This tests that rewriting state works correctly. -// This creates a small room with a create/join/name state, then replays it -// with a new room name. We expect the output events to contain the original events, -// followed by a single OutputNewRoomEvent with RewritesState set to true with the -// rewritten state events (with the 2nd room name). -func TestOutputRewritesState(t *testing.T) { - roomID := "!foo:" + string(testOrigin) - alice := "@alice:" + string(testOrigin) - emptyKey := "" - originalEvents := mustCreateEvents(t, gomatrixserverlib.RoomVersionV6, []fledglingEvent{ - { - RoomID: roomID, - Sender: alice, - Content: map[string]interface{}{ - "creator": alice, - "room_version": "6", - }, - StateKey: &emptyKey, - Type: gomatrixserverlib.MRoomCreate, - }, - { - RoomID: roomID, - Sender: alice, - Content: map[string]interface{}{ - "membership": "join", - }, - StateKey: &alice, - Type: gomatrixserverlib.MRoomMember, - }, - { - RoomID: roomID, - Sender: alice, - Content: map[string]interface{}{ - "body": "hello world", - }, - StateKey: nil, - Type: "m.room.message", - }, - { - RoomID: roomID, - Sender: alice, - Content: map[string]interface{}{ - "name": "Room Name", - }, - StateKey: &emptyKey, - Type: "m.room.name", - }, - }) - rewriteEvents := mustCreateEvents(t, gomatrixserverlib.RoomVersionV6, []fledglingEvent{ - { - RoomID: roomID, - Sender: alice, - Content: map[string]interface{}{ - "creator": alice, - }, - StateKey: &emptyKey, - Type: gomatrixserverlib.MRoomCreate, - }, - { - RoomID: roomID, - Sender: alice, - Content: map[string]interface{}{ - "membership": "join", - }, - StateKey: &alice, - Type: gomatrixserverlib.MRoomMember, - }, - { - RoomID: roomID, - Sender: alice, - Content: map[string]interface{}{ - "name": "Room Name 2", - }, - StateKey: &emptyKey, - Type: "m.room.name", - }, - { - RoomID: roomID, - Sender: alice, - Content: map[string]interface{}{ - "body": "hello world 2", - }, - StateKey: nil, - Type: "m.room.message", - }, - }) - deleteDatabase() - rsAPI, producer := mustCreateRoomserverAPI(t) - defer deleteDatabase() - err := api.SendEvents(context.Background(), rsAPI, api.KindNew, originalEvents, testOrigin, nil) - if err != nil { - t.Fatalf("failed to send original events: %s", err) - } - // assert we got them produced, this is just a sanity check and isn't the intention of this test - if len(producer.producedMessages) != len(originalEvents) { - t.Fatalf("SendEvents didn't result in same number of produced output events: got %d want %d", len(producer.producedMessages), len(originalEvents)) - } - producer.producedMessages = nil // we aren't actually interested in these events, just the rewrite ones - - var inputEvents []api.InputRoomEvent - // slowly build up the state IDs again, we're basically telling the roomserver what to store as a snapshot - var stateIDs []string - // skip the last event, we'll use this to tie together the rewrite as the KindNew event - for i := 0; i < len(rewriteEvents)-1; i++ { - ev := rewriteEvents[i] - inputEvents = append(inputEvents, api.InputRoomEvent{ - Kind: api.KindOutlier, - Event: ev, - AuthEventIDs: ev.AuthEventIDs(), - HasState: true, - StateEventIDs: stateIDs, - }) - if ev.StateKey() != nil { - stateIDs = append(stateIDs, ev.EventID()) - } - } - lastEv := rewriteEvents[len(rewriteEvents)-1] - inputEvents = append(inputEvents, api.InputRoomEvent{ - Kind: api.KindNew, - Event: lastEv, - AuthEventIDs: lastEv.AuthEventIDs(), - HasState: true, - StateEventIDs: stateIDs, - }) - if err := api.SendInputRoomEvents(context.Background(), rsAPI, inputEvents); err != nil { - t.Fatalf("SendInputRoomEvents returned error for rewrite events: %s", err) - } - // we should just have one output event with the entire state of the room in it - if len(producer.producedMessages) != 1 { - t.Fatalf("Rewritten events got output, want only 1 got %d", len(producer.producedMessages)) - } - outputEvent := producer.producedMessages[len(producer.producedMessages)-1] - if !outputEvent.NewRoomEvent.RewritesState { - t.Errorf("RewritesState flag not set on output event") - } - if !reflect.DeepEqual(stateIDs, outputEvent.NewRoomEvent.AddsStateEventIDs) { - t.Errorf("Output event is missing room state event IDs, got %v want %v", outputEvent.NewRoomEvent.AddsStateEventIDs, stateIDs) - } - if !bytes.Equal(outputEvent.NewRoomEvent.Event.JSON(), lastEv.JSON()) { - t.Errorf( - "Output event isn't the latest KindNew event:\ngot %s\nwant %s", - string(outputEvent.NewRoomEvent.Event.JSON()), - string(lastEv.JSON()), - ) - } - if len(outputEvent.NewRoomEvent.AddStateEvents) != len(stateIDs) { - t.Errorf("Output event is missing room state events themselves, got %d want %d", len(outputEvent.NewRoomEvent.AddStateEvents), len(stateIDs)) - } - // make sure the state got overwritten, check the room name - hasRoomName := false - for _, ev := range outputEvent.NewRoomEvent.AddStateEvents { - if ev.Type() == "m.room.name" { - hasRoomName = string(ev.Content()) == `{"name":"Room Name 2"}` - } - } - if !hasRoomName { - t.Errorf("Output event did not overwrite room state") - } -} diff --git a/setup/jetstream/nats.go b/setup/jetstream/nats.go index f074f5f20..43cfc98b9 100644 --- a/setup/jetstream/nats.go +++ b/setup/jetstream/nats.go @@ -18,7 +18,7 @@ import ( var natsServer *natsserver.Server var natsServerMutex sync.Mutex -func SetupConsumerProducer(cfg *config.JetStream) (sarama.Consumer, sarama.SyncProducer) { +func Prepare(cfg *config.JetStream) (nats.JetStreamContext, sarama.Consumer, sarama.SyncProducer) { // check if we need an in-process NATS Server if len(cfg.Addresses) != 0 { return setupNATS(cfg, nil) @@ -51,30 +51,30 @@ func SetupConsumerProducer(cfg *config.JetStream) (sarama.Consumer, sarama.SyncP return setupNATS(cfg, nc) } -func setupNATS(cfg *config.JetStream, nc *natsclient.Conn) (sarama.Consumer, sarama.SyncProducer) { +func setupNATS(cfg *config.JetStream, nc *natsclient.Conn) (nats.JetStreamContext, sarama.Consumer, sarama.SyncProducer) { if nc == nil { var err error nc, err = nats.Connect(strings.Join(cfg.Addresses, ",")) if err != nil { logrus.WithError(err).Panic("Unable to connect to NATS") - return nil, nil + return nil, nil, nil } } s, err := nc.JetStream() if err != nil { logrus.WithError(err).Panic("Unable to get JetStream context") - return nil, nil + return nil, nil, nil } for _, stream := range streams { - stream.Name = cfg.TopicFor(stream.Name) - info, err := s.StreamInfo(stream.Name) + name := cfg.TopicFor(stream.Name) + info, err := s.StreamInfo(name) if err != nil && err != natsclient.ErrStreamNotFound { logrus.WithError(err).Fatal("Unable to get stream info") } if info == nil { - stream.Subjects = []string{stream.Name} + 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 { @@ -82,12 +82,12 @@ func setupNATS(cfg *config.JetStream, nc *natsclient.Conn) (sarama.Consumer, sar } if _, err = s.AddStream(stream); err != nil { - logrus.WithError(err).WithField("stream", stream.Name).Fatal("Unable to add stream") + logrus.WithError(err).WithField("stream", name).Fatal("Unable to add stream") } } } consumer := saramajs.NewJetStreamConsumer(nc, s, "") producer := saramajs.NewJetStreamProducer(nc, s, "") - return consumer, producer + return s, consumer, producer } diff --git a/setup/jetstream/streams.go b/setup/jetstream/streams.go index 326e62a93..0fd31082c 100644 --- a/setup/jetstream/streams.go +++ b/setup/jetstream/streams.go @@ -6,7 +6,13 @@ import ( "github.com/nats-io/nats.go" ) +const ( + UserID = "user_id" + RoomID = "room_id" +) + var ( + InputRoomEvent = "InputRoomEvent" OutputRoomEvent = "OutputRoomEvent" OutputSendToDeviceEvent = "OutputSendToDeviceEvent" OutputKeyChangeEvent = "OutputKeyChangeEvent" @@ -16,6 +22,11 @@ var ( ) var streams = []*nats.StreamConfig{ + { + Name: InputRoomEvent, + Retention: nats.InterestPolicy, + Storage: nats.FileStorage, + }, { Name: OutputRoomEvent, Retention: nats.InterestPolicy, diff --git a/setup/mscs/msc2836/msc2836.go b/setup/mscs/msc2836/msc2836.go index a538299dc..5e7896481 100644 --- a/setup/mscs/msc2836/msc2836.go +++ b/setup/mscs/msc2836/msc2836.go @@ -649,7 +649,7 @@ func (rc *reqCtx) injectResponseToRoomserver(res *gomatrixserverlib.MSC2836Event }) } // we've got the data by this point so use a background context - err = roomserver.SendInputRoomEvents(context.Background(), rc.rsAPI, ires) + err = roomserver.SendInputRoomEvents(context.Background(), rc.rsAPI, ires, false) if err != nil { util.GetLogger(rc.ctx).WithError(err).Error("failed to inject MSC2836EventRelationshipsResponse into the roomserver") } diff --git a/syncapi/consumers/clientapi.go b/syncapi/consumers/clientapi.go index a50e9c8f9..06756fb56 100644 --- a/syncapi/consumers/clientapi.go +++ b/syncapi/consumers/clientapi.go @@ -18,9 +18,7 @@ import ( "context" "encoding/json" - "github.com/Shopify/sarama" "github.com/getsentry/sentry-go" - "github.com/matrix-org/dendrite/internal" "github.com/matrix-org/dendrite/internal/eventutil" "github.com/matrix-org/dendrite/setup/config" "github.com/matrix-org/dendrite/setup/jetstream" @@ -28,60 +26,56 @@ import ( "github.com/matrix-org/dendrite/syncapi/notifier" "github.com/matrix-org/dendrite/syncapi/storage" "github.com/matrix-org/dendrite/syncapi/types" + "github.com/nats-io/nats.go" log "github.com/sirupsen/logrus" ) // OutputClientDataConsumer consumes events that originated in the client API server. type OutputClientDataConsumer struct { - clientAPIConsumer *internal.ContinualConsumer - db storage.Database - stream types.StreamProvider - notifier *notifier.Notifier + jetstream nats.JetStreamContext + topic string + db storage.Database + stream types.StreamProvider + notifier *notifier.Notifier } // NewOutputClientDataConsumer creates a new OutputClientData consumer. Call Start() to begin consuming from room servers. func NewOutputClientDataConsumer( process *process.ProcessContext, cfg *config.SyncAPI, - kafkaConsumer sarama.Consumer, + js nats.JetStreamContext, store storage.Database, notifier *notifier.Notifier, stream types.StreamProvider, ) *OutputClientDataConsumer { - consumer := internal.ContinualConsumer{ - Process: process, - ComponentName: "syncapi/clientapi", - Topic: string(cfg.Matrix.JetStream.TopicFor(jetstream.OutputClientData)), - Consumer: kafkaConsumer, - PartitionStore: store, + return &OutputClientDataConsumer{ + jetstream: js, + topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputClientData), + db: store, + notifier: notifier, + stream: stream, } - s := &OutputClientDataConsumer{ - clientAPIConsumer: &consumer, - db: store, - notifier: notifier, - stream: stream, - } - consumer.ProcessMessage = s.onMessage - - return s } // Start consuming from room servers func (s *OutputClientDataConsumer) Start() error { - return s.clientAPIConsumer.Start() + _, err := s.jetstream.Subscribe(s.topic, s.onMessage) + return err } // onMessage is called when the sync server receives a new event from the client API server output log. // It is not safe for this function to be called from multiple goroutines, or else the // sync stream position may race and be incorrectly calculated. -func (s *OutputClientDataConsumer) onMessage(msg *sarama.ConsumerMessage) error { +func (s *OutputClientDataConsumer) onMessage(msg *nats.Msg) { // Parse out the event JSON + userID := msg.Header.Get(jetstream.UserID) var output eventutil.AccountData - if err := json.Unmarshal(msg.Value, &output); err != nil { + if err := json.Unmarshal(msg.Data, &output); err != nil { // If the message was invalid, log it and move on to the next message in the stream log.WithError(err).Errorf("client API server output log: message parse failure") sentry.CaptureException(err) - return nil + _ = msg.Ack() + return } log.WithFields(log.Fields{ @@ -90,7 +84,7 @@ func (s *OutputClientDataConsumer) onMessage(msg *sarama.ConsumerMessage) error }).Info("received data from client API server") streamPos, err := s.db.UpsertAccountData( - context.TODO(), string(msg.Key), output.RoomID, output.Type, + context.TODO(), userID, output.RoomID, output.Type, ) if err != nil { sentry.CaptureException(err) @@ -102,7 +96,7 @@ func (s *OutputClientDataConsumer) onMessage(msg *sarama.ConsumerMessage) error } s.stream.Advance(streamPos) - s.notifier.OnNewAccountData(string(msg.Key), types.StreamingToken{AccountDataPosition: streamPos}) + s.notifier.OnNewAccountData(userID, types.StreamingToken{AccountDataPosition: streamPos}) - return nil + _ = msg.Ack() } diff --git a/syncapi/consumers/eduserver_receipts.go b/syncapi/consumers/eduserver_receipts.go index 2b25985f9..99189a785 100644 --- a/syncapi/consumers/eduserver_receipts.go +++ b/syncapi/consumers/eduserver_receipts.go @@ -18,25 +18,25 @@ import ( "context" "encoding/json" - "github.com/Shopify/sarama" "github.com/getsentry/sentry-go" "github.com/matrix-org/dendrite/eduserver/api" - "github.com/matrix-org/dendrite/internal" "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/notifier" "github.com/matrix-org/dendrite/syncapi/storage" "github.com/matrix-org/dendrite/syncapi/types" + "github.com/nats-io/nats.go" log "github.com/sirupsen/logrus" ) // OutputReceiptEventConsumer consumes events that originated in the EDU server. type OutputReceiptEventConsumer struct { - receiptConsumer *internal.ContinualConsumer - db storage.Database - stream types.StreamProvider - notifier *notifier.Notifier + jetstream nats.JetStreamContext + topic string + db storage.Database + stream types.StreamProvider + notifier *notifier.Notifier } // NewOutputReceiptEventConsumer creates a new OutputReceiptEventConsumer. @@ -44,44 +44,34 @@ type OutputReceiptEventConsumer struct { func NewOutputReceiptEventConsumer( process *process.ProcessContext, cfg *config.SyncAPI, - kafkaConsumer sarama.Consumer, + js nats.JetStreamContext, store storage.Database, notifier *notifier.Notifier, stream types.StreamProvider, ) *OutputReceiptEventConsumer { - - consumer := internal.ContinualConsumer{ - Process: process, - ComponentName: "syncapi/eduserver/receipt", - Topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputReceiptEvent), - Consumer: kafkaConsumer, - PartitionStore: store, + return &OutputReceiptEventConsumer{ + jetstream: js, + topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputReceiptEvent), + db: store, + notifier: notifier, + stream: stream, } - - s := &OutputReceiptEventConsumer{ - receiptConsumer: &consumer, - db: store, - notifier: notifier, - stream: stream, - } - - consumer.ProcessMessage = s.onMessage - - return s } // Start consuming from EDU api func (s *OutputReceiptEventConsumer) Start() error { - return s.receiptConsumer.Start() + _, err := s.jetstream.Subscribe(s.topic, s.onMessage) + return err } -func (s *OutputReceiptEventConsumer) onMessage(msg *sarama.ConsumerMessage) error { +func (s *OutputReceiptEventConsumer) onMessage(msg *nats.Msg) { var output api.OutputReceiptEvent - if err := json.Unmarshal(msg.Value, &output); err != nil { + if err := json.Unmarshal(msg.Data, &output); err != nil { // If the message was invalid, log it and move on to the next message in the stream log.WithError(err).Errorf("EDU server output log: message parse failure") sentry.CaptureException(err) - return nil + _ = msg.Ack() + return } streamPos, err := s.db.StoreReceipt( @@ -94,11 +84,11 @@ func (s *OutputReceiptEventConsumer) onMessage(msg *sarama.ConsumerMessage) erro ) if err != nil { sentry.CaptureException(err) - return err + return } s.stream.Advance(streamPos) s.notifier.OnNewReceipt(output.RoomID, types.StreamingToken{ReceiptPosition: streamPos}) - return nil + _ = msg.Ack() } diff --git a/syncapi/consumers/eduserver_sendtodevice.go b/syncapi/consumers/eduserver_sendtodevice.go index 4de8eeb77..9ffd572e6 100644 --- a/syncapi/consumers/eduserver_sendtodevice.go +++ b/syncapi/consumers/eduserver_sendtodevice.go @@ -18,10 +18,8 @@ import ( "context" "encoding/json" - "github.com/Shopify/sarama" "github.com/getsentry/sentry-go" "github.com/matrix-org/dendrite/eduserver/api" - "github.com/matrix-org/dendrite/internal" "github.com/matrix-org/dendrite/setup/config" "github.com/matrix-org/dendrite/setup/jetstream" "github.com/matrix-org/dendrite/setup/process" @@ -30,16 +28,18 @@ import ( "github.com/matrix-org/dendrite/syncapi/types" "github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/util" + "github.com/nats-io/nats.go" log "github.com/sirupsen/logrus" ) // OutputSendToDeviceEventConsumer consumes events that originated in the EDU server. type OutputSendToDeviceEventConsumer struct { - sendToDeviceConsumer *internal.ContinualConsumer - db storage.Database - serverName gomatrixserverlib.ServerName // our server name - stream types.StreamProvider - notifier *notifier.Notifier + jetstream nats.JetStreamContext + topic string + db storage.Database + serverName gomatrixserverlib.ServerName // our server name + stream types.StreamProvider + notifier *notifier.Notifier } // NewOutputSendToDeviceEventConsumer creates a new OutputSendToDeviceEventConsumer. @@ -47,54 +47,46 @@ type OutputSendToDeviceEventConsumer struct { func NewOutputSendToDeviceEventConsumer( process *process.ProcessContext, cfg *config.SyncAPI, - kafkaConsumer sarama.Consumer, + js nats.JetStreamContext, store storage.Database, notifier *notifier.Notifier, stream types.StreamProvider, ) *OutputSendToDeviceEventConsumer { - - consumer := internal.ContinualConsumer{ - Process: process, - ComponentName: "syncapi/eduserver/sendtodevice", - Topic: string(cfg.Matrix.JetStream.TopicFor(jetstream.OutputSendToDeviceEvent)), - Consumer: kafkaConsumer, - PartitionStore: store, + return &OutputSendToDeviceEventConsumer{ + jetstream: js, + topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputSendToDeviceEvent), + db: store, + serverName: cfg.Matrix.ServerName, + notifier: notifier, + stream: stream, } - - s := &OutputSendToDeviceEventConsumer{ - sendToDeviceConsumer: &consumer, - db: store, - serverName: cfg.Matrix.ServerName, - notifier: notifier, - stream: stream, - } - - consumer.ProcessMessage = s.onMessage - - return s } // Start consuming from EDU api func (s *OutputSendToDeviceEventConsumer) Start() error { - return s.sendToDeviceConsumer.Start() + _, err := s.jetstream.Subscribe(s.topic, s.onMessage) + return err } -func (s *OutputSendToDeviceEventConsumer) onMessage(msg *sarama.ConsumerMessage) error { +func (s *OutputSendToDeviceEventConsumer) onMessage(msg *nats.Msg) { var output api.OutputSendToDeviceEvent - if err := json.Unmarshal(msg.Value, &output); err != nil { + if err := json.Unmarshal(msg.Data, &output); err != nil { // If the message was invalid, log it and move on to the next message in the stream log.WithError(err).Errorf("EDU server output log: message parse failure") sentry.CaptureException(err) - return err + _ = msg.Ack() + return } _, domain, err := gomatrixserverlib.SplitID('@', output.UserID) if err != nil { sentry.CaptureException(err) - return err + _ = msg.Ack() + return } if domain != s.serverName { - return nil + _ = msg.Ack() + return } util.GetLogger(context.TODO()).WithFields(log.Fields{ @@ -110,7 +102,7 @@ func (s *OutputSendToDeviceEventConsumer) onMessage(msg *sarama.ConsumerMessage) if err != nil { sentry.CaptureException(err) log.WithError(err).Errorf("failed to store send-to-device message") - return err + return } s.stream.Advance(streamPos) @@ -120,5 +112,5 @@ func (s *OutputSendToDeviceEventConsumer) onMessage(msg *sarama.ConsumerMessage) types.StreamingToken{SendToDevicePosition: streamPos}, ) - return nil + _ = msg.Ack() } diff --git a/syncapi/consumers/eduserver_typing.go b/syncapi/consumers/eduserver_typing.go index 343f777e1..b69293c6d 100644 --- a/syncapi/consumers/eduserver_typing.go +++ b/syncapi/consumers/eduserver_typing.go @@ -17,26 +17,26 @@ package consumers import ( "encoding/json" - "github.com/Shopify/sarama" "github.com/getsentry/sentry-go" "github.com/matrix-org/dendrite/eduserver/api" "github.com/matrix-org/dendrite/eduserver/cache" - "github.com/matrix-org/dendrite/internal" "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/notifier" "github.com/matrix-org/dendrite/syncapi/storage" "github.com/matrix-org/dendrite/syncapi/types" + "github.com/nats-io/nats.go" log "github.com/sirupsen/logrus" ) // OutputTypingEventConsumer consumes events that originated in the EDU server. type OutputTypingEventConsumer struct { - typingConsumer *internal.ContinualConsumer - eduCache *cache.EDUCache - stream types.StreamProvider - notifier *notifier.Notifier + jetstream nats.JetStreamContext + topic string + eduCache *cache.EDUCache + stream types.StreamProvider + notifier *notifier.Notifier } // NewOutputTypingEventConsumer creates a new OutputTypingEventConsumer. @@ -44,49 +44,35 @@ type OutputTypingEventConsumer struct { func NewOutputTypingEventConsumer( process *process.ProcessContext, cfg *config.SyncAPI, - kafkaConsumer sarama.Consumer, + js nats.JetStreamContext, store storage.Database, eduCache *cache.EDUCache, notifier *notifier.Notifier, stream types.StreamProvider, ) *OutputTypingEventConsumer { - - consumer := internal.ContinualConsumer{ - Process: process, - ComponentName: "syncapi/eduserver/typing", - Topic: string(cfg.Matrix.JetStream.TopicFor(jetstream.OutputTypingEvent)), - Consumer: kafkaConsumer, - PartitionStore: store, + return &OutputTypingEventConsumer{ + jetstream: js, + topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputTypingEvent), + eduCache: eduCache, + notifier: notifier, + stream: stream, } - - s := &OutputTypingEventConsumer{ - typingConsumer: &consumer, - eduCache: eduCache, - notifier: notifier, - stream: stream, - } - - consumer.ProcessMessage = s.onMessage - - return s } // Start consuming from EDU api func (s *OutputTypingEventConsumer) Start() error { - s.eduCache.SetTimeoutCallback(func(userID, roomID string, latestSyncPosition int64) { - pos := types.StreamPosition(latestSyncPosition) - s.notifier.OnNewTyping(roomID, types.StreamingToken{TypingPosition: pos}) - }) - return s.typingConsumer.Start() + _, err := s.jetstream.Subscribe(s.topic, s.onMessage) + return err } -func (s *OutputTypingEventConsumer) onMessage(msg *sarama.ConsumerMessage) error { +func (s *OutputTypingEventConsumer) onMessage(msg *nats.Msg) { var output api.OutputTypingEvent - if err := json.Unmarshal(msg.Value, &output); err != nil { + if err := json.Unmarshal(msg.Data, &output); err != nil { // If the message was invalid, log it and move on to the next message in the stream log.WithError(err).Errorf("EDU server output log: message parse failure") sentry.CaptureException(err) - return nil + _ = msg.Ack() + return } log.WithFields(log.Fields{ @@ -110,5 +96,5 @@ func (s *OutputTypingEventConsumer) onMessage(msg *sarama.ConsumerMessage) error s.stream.Advance(typingPos) s.notifier.OnNewTyping(output.Event.RoomID, types.StreamingToken{TypingPosition: typingPos}) - return nil + _ = msg.Ack() } diff --git a/syncapi/consumers/keychange.go b/syncapi/consumers/keychange.go index dfedc6409..af7162f34 100644 --- a/syncapi/consumers/keychange.go +++ b/syncapi/consumers/keychange.go @@ -125,6 +125,9 @@ func (s *OutputKeyChangeEventConsumer) onMessage(msg *sarama.ConsumerMessage) er } func (s *OutputKeyChangeEventConsumer) onDeviceKeyMessage(m api.DeviceMessage, offset int64, partition int32) error { + if m.DeviceKeys == nil { + return nil + } output := m.DeviceKeys // work out who we need to notify about the new key var queryRes roomserverAPI.QuerySharedUsersResponse diff --git a/syncapi/consumers/roomserver.go b/syncapi/consumers/roomserver.go index 7ddf2c092..e03d874b7 100644 --- a/syncapi/consumers/roomserver.go +++ b/syncapi/consumers/roomserver.go @@ -19,9 +19,7 @@ import ( "encoding/json" "fmt" - "github.com/Shopify/sarama" "github.com/getsentry/sentry-go" - "github.com/matrix-org/dendrite/internal" "github.com/matrix-org/dendrite/roomserver/api" "github.com/matrix-org/dendrite/setup/config" "github.com/matrix-org/dendrite/setup/jetstream" @@ -30,6 +28,7 @@ import ( "github.com/matrix-org/dendrite/syncapi/storage" "github.com/matrix-org/dendrite/syncapi/types" "github.com/matrix-org/gomatrixserverlib" + "github.com/nats-io/nats.go" log "github.com/sirupsen/logrus" ) @@ -37,7 +36,8 @@ import ( type OutputRoomEventConsumer struct { cfg *config.SyncAPI rsAPI api.RoomserverInternalAPI - rsConsumer *internal.ContinualConsumer + jetstream nats.JetStreamContext + topic string db storage.Database pduStream types.StreamProvider inviteStream types.StreamProvider @@ -48,50 +48,43 @@ type OutputRoomEventConsumer struct { func NewOutputRoomEventConsumer( process *process.ProcessContext, cfg *config.SyncAPI, - kafkaConsumer sarama.Consumer, + js nats.JetStreamContext, store storage.Database, notifier *notifier.Notifier, pduStream types.StreamProvider, inviteStream types.StreamProvider, rsAPI api.RoomserverInternalAPI, ) *OutputRoomEventConsumer { - - consumer := internal.ContinualConsumer{ - Process: process, - ComponentName: "syncapi/roomserver", - Topic: string(cfg.Matrix.JetStream.TopicFor(jetstream.OutputRoomEvent)), - Consumer: kafkaConsumer, - PartitionStore: store, - } - s := &OutputRoomEventConsumer{ + return &OutputRoomEventConsumer{ cfg: cfg, - rsConsumer: &consumer, + jetstream: js, + topic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputRoomEvent), db: store, notifier: notifier, pduStream: pduStream, inviteStream: inviteStream, rsAPI: rsAPI, } - consumer.ProcessMessage = s.onMessage - - return s } // Start consuming from room servers func (s *OutputRoomEventConsumer) Start() error { - return s.rsConsumer.Start() + _, err := s.jetstream.Subscribe(s.topic, s.onMessage) + return err } // onMessage is called when the sync server receives a new event from the room server output log. // It is not safe for this function to be called from multiple goroutines, or else the // sync stream position may race and be incorrectly calculated. -func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error { +func (s *OutputRoomEventConsumer) onMessage(msg *nats.Msg) { // Parse out the event JSON + var err error var output api.OutputEvent - if err := json.Unmarshal(msg.Value, &output); err != nil { + if err = json.Unmarshal(msg.Data, &output); err != nil { // If the message was invalid, log it and move on to the next message in the stream log.WithError(err).Errorf("roomserver output log: message parse failure") - return nil + _ = msg.Ack() + return } switch output.Type { @@ -103,28 +96,36 @@ func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error { // in the special case where the event redacts itself, just pass the message through because // we will never see the other part of the pair if event.Redacts() != event.EventID() { - return nil + _ = msg.Ack() + return } } - return s.onNewRoomEvent(context.TODO(), *output.NewRoomEvent) + s.onNewRoomEvent(context.TODO(), *output.NewRoomEvent) case api.OutputTypeOldRoomEvent: - return s.onOldRoomEvent(context.TODO(), *output.OldRoomEvent) + s.onOldRoomEvent(context.TODO(), *output.OldRoomEvent) case api.OutputTypeNewInviteEvent: - return s.onNewInviteEvent(context.TODO(), *output.NewInviteEvent) + s.onNewInviteEvent(context.TODO(), *output.NewInviteEvent) case api.OutputTypeRetireInviteEvent: - return s.onRetireInviteEvent(context.TODO(), *output.RetireInviteEvent) + s.onRetireInviteEvent(context.TODO(), *output.RetireInviteEvent) case api.OutputTypeNewPeek: - return s.onNewPeek(context.TODO(), *output.NewPeek) + s.onNewPeek(context.TODO(), *output.NewPeek) case api.OutputTypeRetirePeek: - return s.onRetirePeek(context.TODO(), *output.RetirePeek) + s.onRetirePeek(context.TODO(), *output.RetirePeek) case api.OutputTypeRedactedEvent: - return s.onRedactEvent(context.TODO(), *output.RedactedEvent) + s.onRedactEvent(context.TODO(), *output.RedactedEvent) default: log.WithField("type", output.Type).Debug( "roomserver output log: ignoring unknown output type", ) - return nil + _ = msg.Ack() } + if err != nil { + log.WithError(err).Error("roomserver output log: failed to process event") + _ = msg.Nak() + return + } + + _ = msg.Ack() } func (s *OutputRoomEventConsumer) onRedactEvent( @@ -276,12 +277,12 @@ func (s *OutputRoomEventConsumer) notifyJoinedPeeks(ctx context.Context, ev *gom func (s *OutputRoomEventConsumer) onNewInviteEvent( ctx context.Context, msg api.OutputNewInviteEvent, -) error { +) { if msg.Event.StateKey() == nil { log.WithFields(log.Fields{ "event": string(msg.Event.JSON()), }).Panicf("roomserver output log: invite has no state key") - return nil + return } pduPos, err := s.db.AddInviteEvent(ctx, msg.Event) if err != nil { @@ -293,18 +294,16 @@ func (s *OutputRoomEventConsumer) onNewInviteEvent( "pdupos": pduPos, log.ErrorKey: err, }).Panicf("roomserver output log: write invite failure") - return nil + return } s.inviteStream.Advance(pduPos) s.notifier.OnNewInvite(types.StreamingToken{InvitePosition: pduPos}, *msg.Event.StateKey()) - - return nil } func (s *OutputRoomEventConsumer) onRetireInviteEvent( ctx context.Context, msg api.OutputRetireInviteEvent, -) error { +) { pduPos, err := s.db.RetireInviteEvent(ctx, msg.EventID) if err != nil { sentry.CaptureException(err) @@ -313,19 +312,17 @@ func (s *OutputRoomEventConsumer) onRetireInviteEvent( "event_id": msg.EventID, log.ErrorKey: err, }).Panicf("roomserver output log: remove invite failure") - return nil + return } // Notify any active sync requests that the invite has been retired. s.inviteStream.Advance(pduPos) s.notifier.OnNewInvite(types.StreamingToken{InvitePosition: pduPos}, msg.TargetUserID) - - return nil } func (s *OutputRoomEventConsumer) onNewPeek( ctx context.Context, msg api.OutputNewPeek, -) error { +) { sp, err := s.db.AddPeek(ctx, msg.RoomID, msg.UserID, msg.DeviceID) if err != nil { sentry.CaptureException(err) @@ -333,7 +330,7 @@ func (s *OutputRoomEventConsumer) onNewPeek( log.WithFields(log.Fields{ log.ErrorKey: err, }).Panicf("roomserver output log: write peek failure") - return nil + return } // tell the notifier about the new peek so it knows to wake up new devices @@ -341,20 +338,18 @@ func (s *OutputRoomEventConsumer) onNewPeek( // index as PDUs, but we should fix this s.pduStream.Advance(sp) s.notifier.OnNewPeek(msg.RoomID, msg.UserID, msg.DeviceID, types.StreamingToken{PDUPosition: sp}) - - return nil } func (s *OutputRoomEventConsumer) onRetirePeek( ctx context.Context, msg api.OutputRetirePeek, -) error { +) { sp, err := s.db.DeletePeek(ctx, msg.RoomID, msg.UserID, msg.DeviceID) if err != nil { // panic rather than continue with an inconsistent database log.WithFields(log.Fields{ log.ErrorKey: err, }).Panicf("roomserver output log: write peek failure") - return nil + return } // tell the notifier about the new peek so it knows to wake up new devices @@ -362,8 +357,6 @@ func (s *OutputRoomEventConsumer) onRetirePeek( // index as PDUs, but we should fix this s.pduStream.Advance(sp) s.notifier.OnRetirePeek(msg.RoomID, msg.UserID, msg.DeviceID, types.StreamingToken{PDUPosition: sp}) - - return nil } func (s *OutputRoomEventConsumer) updateStateEvent(event *gomatrixserverlib.HeaderedEvent) (*gomatrixserverlib.HeaderedEvent, error) { diff --git a/syncapi/syncapi.go b/syncapi/syncapi.go index 16e222cb0..39bc233ae 100644 --- a/syncapi/syncapi.go +++ b/syncapi/syncapi.go @@ -48,7 +48,7 @@ func AddPublicRoutes( federation *gomatrixserverlib.FederationClient, cfg *config.SyncAPI, ) { - consumer, _ := jetstream.SetupConsumerProducer(&cfg.Matrix.JetStream) + js, consumer, _ := jetstream.Prepare(&cfg.Matrix.JetStream) syncDB, err := storage.NewSyncServerDatasource(&cfg.Database) if err != nil { @@ -65,15 +65,16 @@ func AddPublicRoutes( requestPool := sync.NewRequestPool(syncDB, cfg, userAPI, keyAPI, rsAPI, streams, notifier) keyChangeConsumer := consumers.NewOutputKeyChangeEventConsumer( - process, cfg.Matrix.ServerName, string(cfg.Matrix.JetStream.TopicFor(jetstream.OutputKeyChangeEvent)), - consumer, keyAPI, rsAPI, syncDB, notifier, streams.DeviceListStreamProvider, + process, cfg.Matrix.ServerName, cfg.Matrix.JetStream.TopicFor(jetstream.OutputKeyChangeEvent), + consumer, keyAPI, rsAPI, syncDB, notifier, + streams.DeviceListStreamProvider, ) if err = keyChangeConsumer.Start(); err != nil { logrus.WithError(err).Panicf("failed to start key change consumer") } roomConsumer := consumers.NewOutputRoomEventConsumer( - process, cfg, consumer, syncDB, notifier, streams.PDUStreamProvider, + process, cfg, js, syncDB, notifier, streams.PDUStreamProvider, streams.InviteStreamProvider, rsAPI, ) if err = roomConsumer.Start(); err != nil { @@ -81,28 +82,28 @@ func AddPublicRoutes( } clientConsumer := consumers.NewOutputClientDataConsumer( - process, cfg, consumer, syncDB, notifier, streams.AccountDataStreamProvider, + process, cfg, js, syncDB, notifier, streams.AccountDataStreamProvider, ) if err = clientConsumer.Start(); err != nil { logrus.WithError(err).Panicf("failed to start client data consumer") } typingConsumer := consumers.NewOutputTypingEventConsumer( - process, cfg, consumer, syncDB, eduCache, notifier, streams.TypingStreamProvider, + process, cfg, js, syncDB, eduCache, notifier, streams.TypingStreamProvider, ) if err = typingConsumer.Start(); err != nil { logrus.WithError(err).Panicf("failed to start typing consumer") } sendToDeviceConsumer := consumers.NewOutputSendToDeviceEventConsumer( - process, cfg, consumer, syncDB, notifier, streams.SendToDeviceStreamProvider, + process, cfg, js, syncDB, notifier, streams.SendToDeviceStreamProvider, ) if err = sendToDeviceConsumer.Start(); err != nil { logrus.WithError(err).Panicf("failed to start send-to-device consumer") } receiptConsumer := consumers.NewOutputReceiptEventConsumer( - process, cfg, consumer, syncDB, notifier, streams.ReceiptStreamProvider, + process, cfg, js, syncDB, notifier, streams.ReceiptStreamProvider, ) if err = receiptConsumer.Start(); err != nil { logrus.WithError(err).Panicf("failed to start receipts consumer") diff --git a/userapi/storage/accounts/interface.go b/userapi/storage/accounts/interface.go index 7af2f15f3..f03b3774c 100644 --- a/userapi/storage/accounts/interface.go +++ b/userapi/storage/accounts/interface.go @@ -20,12 +20,10 @@ import ( "errors" "github.com/matrix-org/dendrite/clientapi/auth/authtypes" - "github.com/matrix-org/dendrite/internal" "github.com/matrix-org/dendrite/userapi/api" ) type Database interface { - internal.PartitionStorer GetAccountByPassword(ctx context.Context, localpart, plaintextPassword string) (*api.Account, error) GetProfileByLocalpart(ctx context.Context, localpart string) (*authtypes.Profile, error) SetPassword(ctx context.Context, localpart string, plaintextPassword string) error