Merge branch 'master' into neilalexander/federationinput

roomserver/internal/api.go

@@ -73,6 +73,7 @@ func NewRoomserverAPI(
 			InputRoomEventTopic:  inputRoomEventTopic,
 			OutputRoomEventTopic: outputRoomEventTopic,
 			JetStream:            consumer,
+			Durable:              cfg.Matrix.JetStream.Durable("RoomserverInputConsumer"),
 			ServerName:           cfg.Matrix.ServerName,
 			ACLs:                 serverACLs,
 		},

roomserver/internal/input/input.go

@@ -45,6 +45,7 @@ var keyContentFields = map[string]string{
 type Inputer struct {
 	DB                   storage.Database
 	JetStream            nats.JetStreamContext
+	Durable              nats.SubOpt
 	ServerName           gomatrixserverlib.ServerName
 	FSAPI                fedapi.FederationInternalAPI
 	KeyRing              gomatrixserverlib.JSONVerifier
@@ -91,6 +92,8 @@ func (r *Inputer) Start() error {
 		// or nak them within a certain amount of time. This stops that from
 		// happening, so we don't end up doing a lot of unnecessary duplicate work.
 		nats.MaxDeliver(0),
+		// Use a durable named consumer.
+		r.Durable,
 	)
 	return err
 }