More refactoring to remove saramajetstream

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 {

appservice/consumers/roomserver.go

@@ -20,20 +20,20 @@ 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
+	jetstream    nats.JetStreamContext
+	topic        string
 	asDB         storage.Database
 	rsAPI        api.RoomserverInternalAPI
 	serverName   string
@@ -45,55 +45,49 @@ 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{
+	return &OutputRoomEventConsumer{
-		Process:        process,
+		jetstream:    js,
-		ComponentName:  "appservice/roomserver",
+		topic:        cfg.Global.JetStream.TopicFor(jetstream.OutputRoomEvent),
-		Topic:          cfg.Global.JetStream.TopicFor(jetstream.OutputRoomEvent),
-		Consumer:       kafkaConsumer,
-		PartitionStore: appserviceDB,
-	}
-	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
+		return
 	}
 
 	if output.Type != api.OutputTypeNewRoomEvent {
-		return nil
+		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")
+	}
 }
 
 // filterRoomserverEvents takes in events and decides whether any of them need

clientapi/clientapi.go

@@ -49,10 +49,10 @@ 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,
+		JetStream: js,
 		Topic:     cfg.Matrix.JetStream.TopicFor(jetstream.OutputClientData),
 	}
 

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
+	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
 }

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),

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{
+	_, err = t.JetStream.PublishMsg(&nats.Msg{
-		Topic: string(t.OutputKeyChangeEventTopic),
+		Subject: t.OutputKeyChangeEventTopic,
-		Key:   sarama.StringEncoder(request.UserID),
+		Header:  nats.Header{},
-		Value: sarama.ByteEncoder(eventJSON),
+		Data:    eventJSON,
-	}
+	})
-
-	_, _, err = t.Producer.SendMessage(m)
 	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{
+	_, err = t.JetStream.PublishMsg(&nats.Msg{
-		Topic: string(t.OutputTypingEventTopic),
+		Subject: t.OutputTypingEventTopic,
-		Key:   sarama.StringEncoder(ite.RoomID),
+		Header:  nats.Header{},
-		Value: sarama.ByteEncoder(eventJSON),
+		Data:    eventJSON,
-	}
+	})
-
-	_, _, err = t.Producer.SendMessage(m)
 	return err
 }
 
@@ -193,14 +189,10 @@ func (t *EDUServerInputAPI) sendToDeviceEvent(ise *api.InputSendToDeviceEvent) e
 			return err
 		}
 
-		m := &sarama.ProducerMessage{
+		if _, err = t.JetStream.PublishMsg(&nats.Msg{
-			Topic: string(t.OutputSendToDeviceEventTopic),
+			Subject: t.OutputTypingEventTopic,
-			Key:   sarama.StringEncoder(ote.UserID),
+			Data:    eventJSON,
-			Value: sarama.ByteEncoder(eventJSON),
+		}); err != nil {
-		}
-
-		_, _, err = t.Producer.SendMessage(m)
-		if 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,
+	_, err = t.JetStream.PublishMsg(&nats.Msg{
-		Key:   sarama.StringEncoder(request.InputReceiptEvent.RoomID + ":" + request.InputReceiptEvent.UserID),
+		Subject: t.OutputReceiptEventTopic,
-		Value: sarama.ByteEncoder(js),
+		Data:    js,
-	}
+	})
-	_, _, err = t.Producer.SendMessage(m)
 	return err
 }

federationsender/consumers/eduserver.go

@@ -17,115 +17,88 @@ 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
+	jetstream         nats.JetStreamContext
-	sendToDeviceConsumer *internal.ContinualConsumer
-	receiptConsumer      *internal.ContinualConsumer
 	db                storage.Database
 	queues            *queue.OutgoingQueues
 	ServerName        gomatrixserverlib.ServerName
-	TypingTopic          string
+	typingTopic       string
-	SendToDeviceTopic    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{
+	return &OutputEDUConsumer{
-		typingConsumer: &internal.ContinualConsumer{
+		jetstream:         js,
-			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,
-		},
 		queues:            queues,
 		db:                store,
 		ServerName:        cfg.Matrix.ServerName,
-		TypingTopic:       cfg.Matrix.JetStream.TopicFor(jetstream.OutputTypingEvent),
+		typingTopic:       cfg.Matrix.JetStream.TopicFor(jetstream.OutputTypingEvent),
-		SendToDeviceTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputSendToDeviceEvent),
+		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 {
+	if _, err := t.jetstream.Subscribe(t.typingTopic, t.onTypingEvent); err != nil {
-		return fmt.Errorf("t.typingConsumer.Start: %w", err)
+		return err
 	}
-	if err := t.sendToDeviceConsumer.Start(); err != nil {
+	if _, err := t.jetstream.Subscribe(t.sendToDeviceTopic, t.onSendToDeviceEvent); err != nil {
-		return fmt.Errorf("t.sendToDeviceConsumer.Start: %w", err)
+		return err
 	}
-	if err := t.receiptConsumer.Start(); err != nil {
+	if _, err := t.jetstream.Subscribe(t.receiptTopic, t.onReceiptEvent); err != nil {
-		return fmt.Errorf("t.receiptConsumer.Start: %w", err)
+		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
+		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
+		return
 	}
 	if originServerName != t.ServerName {
 		log.WithField("other_server", originServerName).Info("Suppressing send-to-device: originated elsewhere")
-		return nil
+		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
+		return
 	}
 
 	// Pack the EDU and marshal it
@@ -144,38 +117,41 @@ 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")
+		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")
+	}
 }
 
 // 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
+		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
+		return
 	}
 	if typingServerName != t.ServerName {
-		log.WithField("other_server", typingServerName).Info("Suppressing typing notif: originated elsewhere")
+		return
-		return nil
 	}
 
 	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 +165,40 @@ 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")
+		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")
+	}
 }
 
 // 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
+		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")
+		log.WithError(err).WithField("user_id", receipt.UserID).Error("failed to extract domain from receipt sender")
-		return nil
+		return
 	}
 	if receiptServerName != t.ServerName {
-		return nil // don't log, very spammy as it logs for each remote receipt
+		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 +223,11 @@ 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")
+		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")
+	}
 }

federationsender/consumers/roomserver.go

@@ -19,16 +19,15 @@ 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"
 )
 
@@ -36,55 +35,48 @@ import (
 type OutputRoomEventConsumer struct {
 	cfg       *config.FederationSender
 	rsAPI     api.RoomserverInternalAPI
-	rsConsumer *internal.ContinualConsumer
+	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{
+	return &OutputRoomEventConsumer{
-		Process:        process,
-		ComponentName:  "federationsender/roomserver",
-		Topic:          string(cfg.Matrix.JetStream.TopicFor(jetstream.OutputRoomEvent)),
-		Consumer:       kafkaConsumer,
-		PartitionStore: store,
-	}
-	s := &OutputRoomEventConsumer{
 		cfg:       cfg,
-		rsConsumer: &consumer,
+		jetstream: js,
 		db:        store,
 		queues:    queues,
 		rsAPI:     rsAPI,
+		topic:     cfg.Matrix.JetStream.TopicFor(jetstream.OutputRoomEvent),
 	}
-	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
+		return
 	}
 
 	switch output.Type {
@@ -93,7 +85,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
 			}
 		}
 
@@ -113,7 +106,7 @@ func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error {
 					log.ErrorKey: err,
 				}).Panicf("roomserver output log: write room event failure")
 			}
-			return nil
+			return
 		}
 	case api.OutputTypeNewInboundPeek:
 		if err := s.processInboundPeek(*output.NewInboundPeek); err != nil {
@@ -121,16 +114,14 @@ func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error {
 				"event":      output.NewInboundPeek,
 				log.ErrorKey: err,
 			}).Panicf("roomserver output log: remote peek event failure")
-			return nil
+			return
 		}
 	default:
 		log.WithField("type", output.Type).Debug(
 			"roomserver output log: ignoring unknown output type",
 		)
-		return nil
+		return
 	}
-
-	return nil
 }
 
 // processInboundPeek starts tracking a new federated inbound peek (replacing the existing one if any)

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,
+		base.ProcessContext, cfg, js, queues, federationSenderDB, rsAPI,
-		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,
 	)

go.mod

@@ -11,7 +11,7 @@ require (
 	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.30.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
@@ -49,7 +49,6 @@ require (
 	github.com/ngrok/sqlmw v0.0.0-20200129213757-d5c93a81bec6
 	github.com/opentracing/opentracing-go v1.2.0
 	github.com/patrickmn/go-cache v2.1.0+incompatible
-	github.com/pierrec/lz4 v2.6.1+incompatible // indirect
 	github.com/pkg/errors v0.9.1
 	github.com/pressly/goose v2.7.0+incompatible
 	github.com/prometheus/client_golang v1.11.0

go.sum

@@ -92,10 +92,12 @@ github.com/S7evinK/saramajetstream v0.0.0-20210709110708-de6efc8c4a32/go.mod h1:
 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/go.mod h1:2QpgD79wpdAESqNQMxNc0KYMkycd4slxGdV3TWSVqrU=
-github.com/Shopify/sarama v1.29.1 h1:wBAacXbYVLmWieEA/0X/JagDdCZ8NVFOfS6l6+2u5S0=
+github.com/Shopify/sarama v1.30.0 h1:TOZL6r37xJBDEMLx4yjB77jxbZYXPaDow08TSK6vIL0=
-github.com/Shopify/sarama v1.29.1/go.mod h1:mdtqvCSg8JOxk8PmpTNGyo6wzd4BMm4QXSfDnTXmgkE=
+github.com/Shopify/sarama v1.30.0/go.mod h1:zujlQQx1kzHsh4jfV1USnptCQrHAEZ2Hk8fTKCulPVs=
 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/Shopify/toxiproxy/v2 v2.1.6-0.20210914104332-15ea381dcdae h1:ePgznFqEG1v3AjMklnK8H7BSc++FDSo7xfK9K7Af+0Y=
+github.com/Shopify/toxiproxy/v2 v2.1.6-0.20210914104332-15ea381dcdae/go.mod h1:/cvHQkZ1fst0EmZnA5dFtiQdWCNCFYzb+uE2vqVgvx0=
 github.com/VividCortex/ewma v1.1.1/go.mod h1:2Tkkvm3sRDVXaiyucHiACn4cqf7DpdyLvmxzcbUokwA=
 github.com/VividCortex/ewma v1.2.0/go.mod h1:nz4BbCtbLyFDeC9SUHbtcT5644juEuWfUAUnGx7j5l4=
 github.com/aead/siphash v1.0.1/go.mod h1:Nywa3cDsYNNK3gaciGTWPwHt0wlpNV15vwmswBAUSII=
@@ -474,8 +476,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 +722,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=
@@ -1393,6 +1397,7 @@ github.com/urfave/cli v0.0.0-20171014202726-7bc6a0acffa5/go.mod h1:70zkFmudgCuE/
 github.com/urfave/cli v1.20.0/go.mod h1:70zkFmudgCuE/ngEzBv17Jvp/497gISqfk5gWijbERA=
 github.com/urfave/cli v1.22.1/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0=
 github.com/urfave/cli v1.22.2/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0=
+github.com/urfave/cli/v2 v2.3.0/go.mod h1:LJmUH05zAU44vOAcrfzZQKsZbVcdbOG8rtL3/XcUArI=
 github.com/urfave/negroni v1.0.0/go.mod h1:Meg73S6kFm/4PpbYdq35yYWoCZ9mS/YSx+lKnmiohz4=
 github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
 github.com/valyala/fasthttp v1.6.0/go.mod h1:FstJa9V+Pj9vQ7OJie2qMHdwemEDaDiSdBnvPM1Su9w=
@@ -1422,6 +1427,9 @@ github.com/willf/bitset v1.1.9/go.mod h1:RjeCKbqT1RxIR/KWY6phxZiaY1IyutSBfGjNPyS
 github.com/willf/bitset v1.1.11-0.20200630133818-d5bec3311243/go.mod h1:RjeCKbqT1RxIR/KWY6phxZiaY1IyutSBfGjNPySAYV4=
 github.com/willf/bitset v1.1.11/go.mod h1:83CECat5yLh5zVOf4P1ErAgKA5UDvKtgyUABdr3+MjI=
 github.com/x-cray/logrus-prefixed-formatter v0.5.2/go.mod h1:2duySbKsL6M18s5GU7VPsoEPHyzalCE06qoARUCeBBE=
+github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
+github.com/xdg-go/scram v1.0.2/go.mod h1:1WAq6h33pAW+iRreB34OORO2Nf7qel3VV3fjBj+hCSs=
+github.com/xdg-go/stringprep v1.0.2/go.mod h1:8F9zXuvzgwmyT5DUm4GUfZGDdT3W+LCvS6+da4O5kxM=
 github.com/xdg/scram v1.0.3/go.mod h1:lB8K/P019DLNhemzwFU4jHLhdvlE6uDZjXFejJXr49I=
 github.com/xdg/stringprep v1.0.3/go.mod h1:Jhud4/sHMO4oL310DaZAKk9ZaJ08SJfe+sJh0HrGL1Y=
 github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=
@@ -1511,6 +1519,7 @@ golang.org/x/crypto v0.0.0-20210421170649-83a5a9bb288b/go.mod h1:T9bdIzuCu7OtxOm
 golang.org/x/crypto v0.0.0-20210506145944-38f3c27a63bf/go.mod h1:P+XmwS30IXTQdn5tA2iutPOUgjI07+tq3H3K9MVA1s8=
 golang.org/x/crypto v0.0.0-20210513164829-c07d793c2f9a/go.mod h1:P+XmwS30IXTQdn5tA2iutPOUgjI07+tq3H3K9MVA1s8=
 golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
+golang.org/x/crypto v0.0.0-20210920023735-84f357641f63/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519 h1:7I4JAnoQBe7ZtJcBaYHi5UtiO8tQHbUSXxL+pnGRANg=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
@@ -1602,7 +1611,7 @@ 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-20210917221730-978cfadd31cf/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=
@@ -1724,9 +1733,11 @@ golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3
 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.5/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=
@@ -1920,6 +1931,7 @@ gopkg.in/yaml.v2 v2.0.0-20170712054546-1be3d31502d6/go.mod h1:JAlM8MvJe8wmxCU4Bl
 gopkg.in/yaml.v2 v2.0.0-20170812160011-eb3733d160e7/go.mod h1:JAlM8MvJe8wmxCU4Bli9HhUf9+ttbYbLASfIpnQbh74=
 gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.3/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.5/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=

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 {

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"
@@ -35,7 +34,6 @@ type RoomserverInternalAPI struct {
 	*perform.Forgetter
 	DB                     storage.Database
 	Cfg                    *config.RoomServer
-	Producer               sarama.SyncProducer
 	Cache                  caching.RoomServerCaches
 	ServerName             gomatrixserverlib.ServerName
 	KeyRing                gomatrixserverlib.JSONVerifier
@@ -47,8 +45,7 @@ type RoomserverInternalAPI struct {
 }
 
 func NewRoomserverAPI(
-	cfg *config.RoomServer, roomserverDB storage.Database,
+	cfg *config.RoomServer, roomserverDB storage.Database, consumer nats.JetStreamContext,
-	consumer nats.JetStreamContext, producer sarama.SyncProducer,
 	inputRoomEventTopic, outputRoomEventTopic string, caches caching.RoomServerCaches,
 	keyRing gomatrixserverlib.JSONVerifier, perspectiveServerNames []gomatrixserverlib.ServerName,
 ) *RoomserverInternalAPI {
@@ -70,8 +67,7 @@ func NewRoomserverAPI(
 			DB:                   roomserverDB,
 			InputRoomEventTopic:  inputRoomEventTopic,
 			OutputRoomEventTopic: outputRoomEventTopic,
-			Consumer:             consumer,
+			JetStream:            consumer,
-			Producer:             producer,
 			ServerName:           cfg.Matrix.ServerName,
 			ACLs:                 serverACLs,
 		},

roomserver/internal/input/input.go

@@ -21,12 +21,12 @@ import (
 	"sync"
 
 	"github.com/Arceliar/phony"
-	"github.com/Shopify/sarama"
 	"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"
@@ -42,8 +42,7 @@ var keyContentFields = map[string]string{
 
 type Inputer struct {
 	DB                   storage.Database
-	Consumer             nats.JetStreamContext
+	JetStream            nats.JetStreamContext
-	Producer             sarama.SyncProducer
 	ServerName           gomatrixserverlib.ServerName
 	ACLs                 *acls.ServerACLs
 	InputRoomEventTopic  string
@@ -53,7 +52,7 @@ type Inputer struct {
 
 // onMessage is called when a new event arrives in the roomserver input stream.
 func (r *Inputer) Start() error {
-	_, err := r.Consumer.Subscribe(
+	_, err := r.JetStream.Subscribe(
 		r.InputRoomEventTopic,
 		func(msg *nats.Msg) {
 			_ = msg.InProgress()
@@ -99,7 +98,7 @@ func (r *Inputer) InputRoomEvents(
 			response.ErrMsg = err.Error()
 			return
 		}
-		if _, err = r.Consumer.PublishMsg(msg); err != nil {
+		if _, err = r.JetStream.PublishMsg(msg); err != nil {
 			response.ErrMsg = err.Error()
 			return
 		}
@@ -109,56 +108,51 @@ func (r *Inputer) InputRoomEvents(
 
 // WriteOutputEvents implements OutputRoomEventWriter
 func (r *Inputer) WriteOutputEvents(roomID string, updates []api.OutputEvent) error {
-	messages := make([]*sarama.ProducerMessage, len(updates))
+	var err error
-	for i := range updates {
+	for _, update := range updates {
-		value, err := json.Marshal(updates[i])
+		msg := &nats.Msg{}
+		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 {
+		if update.NewRoomEvent != nil {
-			eventType := updates[i].NewRoomEvent.Event.Type()
+			eventType := update.NewRoomEvent.Event.Type()
 			logger = logger.WithFields(log.Fields{
 				"event_type":     eventType,
-				"event_id":       updates[i].NewRoomEvent.Event.EventID(),
+				"event_id":       update.NewRoomEvent.Event.EventID(),
-				"adds_state":     len(updates[i].NewRoomEvent.AddsStateEventIDs),
+				"adds_state":     len(update.NewRoomEvent.AddsStateEventIDs),
-				"removes_state":  len(updates[i].NewRoomEvent.RemovesStateEventIDs),
+				"removes_state":  len(update.NewRoomEvent.RemovesStateEventIDs),
-				"send_as_server": updates[i].NewRoomEvent.SendAsServer,
+				"send_as_server": update.NewRoomEvent.SendAsServer,
-				"sender":         updates[i].NewRoomEvent.Event.Sender(),
+				"sender":         update.NewRoomEvent.Event.Sender(),
 			})
-			if updates[i].NewRoomEvent.Event.StateKey() != nil {
+			if update.NewRoomEvent.Event.StateKey() != nil {
-				logger = logger.WithField("state_key", *updates[i].NewRoomEvent.Event.StateKey())
+				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("") {
+			if eventType == "m.room.server_acl" && update.NewRoomEvent.Event.StateKeyEquals("") {
-				ev := updates[i].NewRoomEvent.Event.Unwrap()
+				ev := update.NewRoomEvent.Event.Unwrap()
 				defer r.ACLs.OnServerACLUpdate(ev)
 			}
 		}
-		logger.Infof("Producing to topic '%s'", r.OutputRoomEventTopic)
+		logger.Tracef("Producing to topic '%s'", r.OutputRoomEventTopic)
-		messages[i] = &sarama.ProducerMessage{
+		if _, err := r.JetStream.PublishMsg(msg); err != nil {
-			Topic: r.OutputRoomEventTopic,
+			logger.WithError(err).Errorf("Failed to produce to topic '%s': %s", r.OutputRoomEventTopic, err)
-			Key:   sarama.StringEncoder(roomID),
+			return err
-			Value: sarama.ByteEncoder(value),
 		}
 	}
-	errs := r.Producer.SendMessages(messages)
+	return nil
-	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
 }
 
 func init() {

roomserver/roomserver.go

@@ -41,8 +41,6 @@ func NewInternalAPI(
 ) api.RoomserverInternalAPI {
 	cfg := &base.Cfg.RoomServer
 
-	js, _, 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,10 @@ func NewInternalAPI(
 		logrus.WithError(err).Panicf("failed to connect to room server db")
 	}
 
+	js, _, _ := jetstream.Prepare(&cfg.Matrix.JetStream)
+
 	return internal.NewRoomserverAPI(
-		cfg, roomserverDB, js, producer,
+		cfg, roomserverDB, js,
 		cfg.Matrix.JetStream.TopicFor(jetstream.InputRoomEvent),
 		cfg.Matrix.JetStream.TopicFor(jetstream.OutputRoomEvent),
 		base.Caches, keyRing, perspectiveServerNames,

roomserver/roomserver_test.go

@@ -1,410 +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,
-		cfg.Global.JetStream.TopicFor(jetstream.InputRoomEvent),
-		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")
-	}
-}

setup/jetstream/nats.go

@@ -18,7 +18,7 @@ import (
 var natsServer *natsserver.Server
 var natsServerMutex sync.Mutex
 
-func SetupConsumerProducer(cfg *config.JetStream) (nats.JetStreamContext, 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)

setup/jetstream/streams.go

@@ -6,6 +6,11 @@ import (
 	"github.com/nats-io/nats.go"
 )
 
+const (
+	UserID = "user_id"
+	RoomID = "room_id"
+)
+
 var (
 	InputRoomEvent          = "InputRoomEvent"
 	OutputRoomEvent         = "OutputRoomEvent"

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,12 +26,14 @@ 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
+	jetstream nats.JetStreamContext
+	topic     string
 	db        storage.Database
 	stream    types.StreamProvider
 	notifier  *notifier.Notifier
@@ -43,45 +43,38 @@ type OutputClientDataConsumer struct {
 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{
+	return &OutputClientDataConsumer{
-		Process:        process,
+		jetstream: js,
-		ComponentName:  "syncapi/clientapi",
+		topic:     cfg.Matrix.JetStream.TopicFor(jetstream.OutputClientData),
-		Topic:          string(cfg.Matrix.JetStream.TopicFor(jetstream.OutputClientData)),
-		Consumer:       kafkaConsumer,
-		PartitionStore: store,
-	}
-	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
+		return
 	}
 
 	log.WithFields(log.Fields{
@@ -90,7 +83,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 +95,5 @@ 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
 }

syncapi/consumers/eduserver_receipts.go

@@ -18,22 +18,22 @@ 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
+	jetstream nats.JetStreamContext
+	topic     string
 	db        storage.Database
 	stream    types.StreamProvider
 	notifier  *notifier.Notifier
@@ -44,44 +44,33 @@ 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 {
-
+	return &OutputReceiptEventConsumer{
-	consumer := internal.ContinualConsumer{
+		jetstream: js,
-		Process:        process,
+		topic:     cfg.Matrix.JetStream.TopicFor(jetstream.OutputReceiptEvent),
-		ComponentName:  "syncapi/eduserver/receipt",
-		Topic:          cfg.Matrix.JetStream.TopicFor(jetstream.OutputReceiptEvent),
-		Consumer:       kafkaConsumer,
-		PartitionStore: store,
-	}
-
-	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
+		return
 	}
 
 	streamPos, err := s.db.StoreReceipt(
@@ -94,11 +83,9 @@ 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
 }

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,12 +28,14 @@ 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
+	jetstream  nats.JetStreamContext
+	topic      string
 	db         storage.Database
 	serverName gomatrixserverlib.ServerName // our server name
 	stream     types.StreamProvider
@@ -47,54 +47,43 @@ 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 {
-
+	return &OutputSendToDeviceEventConsumer{
-	consumer := internal.ContinualConsumer{
+		jetstream:  js,
-		Process:        process,
+		topic:      cfg.Matrix.JetStream.TopicFor(jetstream.OutputSendToDeviceEvent),
-		ComponentName:  "syncapi/eduserver/sendtodevice",
-		Topic:          string(cfg.Matrix.JetStream.TopicFor(jetstream.OutputSendToDeviceEvent)),
-		Consumer:       kafkaConsumer,
-		PartitionStore: store,
-	}
-
-	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
+		return
 	}
 
 	_, domain, err := gomatrixserverlib.SplitID('@', output.UserID)
 	if err != nil {
 		sentry.CaptureException(err)
-		return err
+		return
 	}
 	if domain != s.serverName {
-		return nil
+		return
 	}
 
 	util.GetLogger(context.TODO()).WithFields(log.Fields{
@@ -110,7 +99,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)
@@ -119,6 +108,4 @@ func (s *OutputSendToDeviceEventConsumer) onMessage(msg *sarama.ConsumerMessage)
 		[]string{output.DeviceID},
 		types.StreamingToken{SendToDevicePosition: streamPos},
 	)
-
-	return nil
 }

syncapi/consumers/eduserver_typing.go

@@ -17,23 +17,22 @@ 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
+	jetstream nats.JetStreamContext
+	topic     string
 	eduCache  *cache.EDUCache
 	stream    types.StreamProvider
 	notifier  *notifier.Notifier
@@ -44,49 +43,33 @@ 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 {
-
+	return &OutputTypingEventConsumer{
-	consumer := internal.ContinualConsumer{
+		jetstream: js,
-		Process:        process,
-		ComponentName:  "syncapi/eduserver/typing",
-		Topic:          string(cfg.Matrix.JetStream.TopicFor(jetstream.OutputTypingEvent)),
-		Consumer:       kafkaConsumer,
-		PartitionStore: store,
-	}
-
-	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) {
+	_, err := s.jetstream.Subscribe(s.topic, s.onMessage)
-		pos := types.StreamPosition(latestSyncPosition)
+	return err
-		s.notifier.OnNewTyping(roomID, types.StreamingToken{TypingPosition: pos})
-	})
-	return s.typingConsumer.Start()
 }
 
-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
+		return
 	}
 
 	log.WithFields(log.Fields{
@@ -109,6 +92,4 @@ func (s *OutputTypingEventConsumer) onMessage(msg *sarama.ConsumerMessage) error
 
 	s.stream.Advance(typingPos)
 	s.notifier.OnNewTyping(output.Event.RoomID, types.StreamingToken{TypingPosition: typingPos})
-
-	return nil
 }

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,42 @@ 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 {
-
+	return &OutputRoomEventConsumer{
-	consumer := internal.ContinualConsumer{
-		Process:        process,
-		ComponentName:  "syncapi/roomserver",
-		Topic:          string(cfg.Matrix.JetStream.TopicFor(jetstream.OutputRoomEvent)),
-		Consumer:       kafkaConsumer,
-		PartitionStore: store,
-	}
-	s := &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
+		return
 	}
 
 	switch output.Type {
@@ -103,27 +95,29 @@ 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
+				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
+	}
+	if err != nil {
+		log.WithError(err).Error("roomserver output log: failed to process event")
 	}
 }
 
@@ -276,12 +270,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 +287,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 +305,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 +323,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 +331,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 +350,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) {

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)),
+		process, cfg.Matrix.ServerName, cfg.Matrix.JetStream.TopicFor(jetstream.OutputKeyChangeEvent),
-		consumer, keyAPI, rsAPI, syncDB, notifier, streams.DeviceListStreamProvider,
+		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")