Merge branch 'nats' into add-nats-support

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

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)

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(

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
 }

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

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

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()
 	}

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()
 	}

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

clientapi/threepid/invites.go

@@ -367,5 +367,6 @@ func emit3PIDInviteEvent(
 		},
 		cfg.Matrix.ServerName,
 		nil,
+		false,
 	)
 }

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

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

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

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

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()
 }

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

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)

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

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

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

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=

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/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

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

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
 }
 

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

roomserver/internal/input/input_fifo.go

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

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

roomserver/roomserver_test.go

@@ -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")
-	}
-}

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
 }

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,

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")
 	}

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()
 }

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()
 }

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()
 }

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()
 }

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

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

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

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