federationsender/roomserver: don't panic while federation is disabled (#1615)

federationsender/consumers/roomserver.go

@@ -94,6 +94,12 @@ func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error {
 		}
 
 		if err := s.processMessage(*output.NewRoomEvent); err != nil {
+			switch err.(type) {
+			case *queue.ErrorFederationDisabled:
+				log.WithField("error", output.Type).Info(
+					err.Error(),
+				)
+			default:
 				// panic rather than continue with an inconsistent database
 				log.WithFields(log.Fields{
 					"event":      string(ev.JSON()),
@@ -101,6 +107,7 @@ func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error {
 					"del":        output.NewRoomEvent.RemovesStateEventIDs,
 					log.ErrorKey: err,
 				}).Panicf("roomserver output log: write room event failure")
+			}
 			return nil
 		}
 	default:

federationsender/queue/queue.go

@@ -122,13 +122,23 @@ func (oqs *OutgoingQueues) getQueue(destination gomatrixserverlib.ServerName) *d
 	return oq
 }
 
+type ErrorFederationDisabled struct {
+	Message string
+}
+
+func (e *ErrorFederationDisabled) Error() string {
+	return e.Message
+}
+
 // SendEvent sends an event to the destinations
 func (oqs *OutgoingQueues) SendEvent(
 	ev *gomatrixserverlib.HeaderedEvent, origin gomatrixserverlib.ServerName,
 	destinations []gomatrixserverlib.ServerName,
 ) error {
 	if oqs.disabled {
-		return fmt.Errorf("federation is disabled")
+		return &ErrorFederationDisabled{
+			Message: "Federation disabled",
+		}
 	}
 	if origin != oqs.origin {
 		// TODO: Support virtual hosting; gh issue #577.
@@ -190,7 +200,9 @@ func (oqs *OutgoingQueues) SendEDU(
 	destinations []gomatrixserverlib.ServerName,
 ) error {
 	if oqs.disabled {
-		return fmt.Errorf("federation is disabled")
+		return &ErrorFederationDisabled{
+			Message: "Federation disabled",
+		}
 	}
 	if origin != oqs.origin {
 		// TODO: Support virtual hosting; gh issue #577.