Message acknowledgements

syncapi/consumers/clientapi.go

@@ -74,6 +74,7 @@ func (s *OutputClientDataConsumer) onMessage(msg *nats.Msg) {
 		// 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)
+		_ = msg.Nak()
 		return
 	}
 
@@ -96,4 +97,6 @@ func (s *OutputClientDataConsumer) onMessage(msg *nats.Msg) {
 
 	s.stream.Advance(streamPos)
 	s.notifier.OnNewAccountData(userID, types.StreamingToken{AccountDataPosition: streamPos})
+
+	_ = msg.Ack()
 }

syncapi/consumers/eduserver_receipts.go

@@ -70,6 +70,7 @@ func (s *OutputReceiptEventConsumer) onMessage(msg *nats.Msg) {
 		// 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)
+		_ = msg.Nak()
 		return
 	}
 
@@ -88,4 +89,6 @@ func (s *OutputReceiptEventConsumer) onMessage(msg *nats.Msg) {
 
 	s.stream.Advance(streamPos)
 	s.notifier.OnNewReceipt(output.RoomID, types.StreamingToken{ReceiptPosition: streamPos})
+
+	_ = msg.Ack()
 }

syncapi/consumers/eduserver_sendtodevice.go

@@ -74,15 +74,18 @@ func (s *OutputSendToDeviceEventConsumer) onMessage(msg *nats.Msg) {
 		// 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)
+		_ = msg.Nak()
 		return
 	}
 
 	_, domain, err := gomatrixserverlib.SplitID('@', output.UserID)
 	if err != nil {
 		sentry.CaptureException(err)
+		_ = msg.Nak()
 		return
 	}
 	if domain != s.serverName {
+		_ = msg.Nak()
 		return
 	}
 
@@ -108,4 +111,6 @@ func (s *OutputSendToDeviceEventConsumer) onMessage(msg *nats.Msg) {
 		[]string{output.DeviceID},
 		types.StreamingToken{SendToDevicePosition: streamPos},
 	)
+
+	_ = msg.Ack()
 }

syncapi/consumers/eduserver_typing.go

@@ -71,6 +71,7 @@ func (s *OutputTypingEventConsumer) onMessage(msg *nats.Msg) {
 		// 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)
+		_ = msg.Nak()
 		return
 	}
 
@@ -94,4 +95,6 @@ func (s *OutputTypingEventConsumer) onMessage(msg *nats.Msg) {
 
 	s.stream.Advance(typingPos)
 	s.notifier.OnNewTyping(output.Event.RoomID, types.StreamingToken{TypingPosition: typingPos})
+
+	_ = msg.Ack()
 }

syncapi/consumers/roomserver.go

@@ -83,6 +83,7 @@ func (s *OutputRoomEventConsumer) onMessage(msg *nats.Msg) {
 	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")
+		_ = msg.Nak()
 		return
 	}
 
@@ -115,10 +116,15 @@ func (s *OutputRoomEventConsumer) onMessage(msg *nats.Msg) {
 		log.WithField("type", output.Type).Debug(
 			"roomserver output log: ignoring unknown output type",
 		)
+		_ = msg.Nak()
 	}
 	if err != nil {
 		log.WithError(err).Error("roomserver output log: failed to process event")
+		_ = msg.Nak()
+		return
 	}
+
+	_ = msg.Ack()
 }
 
 func (s *OutputRoomEventConsumer) onRedactEvent(