diff --git a/federationapi/consumers/keychange.go b/federationapi/consumers/keychange.go index f3314bc9..67dfdc1d 100644 --- a/federationapi/consumers/keychange.go +++ b/federationapi/consumers/keychange.go @@ -18,6 +18,11 @@ import ( "context" "encoding/json" + "github.com/getsentry/sentry-go" + "github.com/matrix-org/gomatrixserverlib" + "github.com/nats-io/nats.go" + "github.com/sirupsen/logrus" + "github.com/matrix-org/dendrite/federationapi/queue" "github.com/matrix-org/dendrite/federationapi/storage" "github.com/matrix-org/dendrite/federationapi/types" @@ -26,9 +31,6 @@ import ( "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/sirupsen/logrus" ) // KeyChangeConsumer consumes events that originate in key server. @@ -78,6 +80,7 @@ func (t *KeyChangeConsumer) onMessage(ctx context.Context, msgs []*nats.Msg) boo msg := msgs[0] // Guaranteed to exist if onMessage is called var m api.DeviceMessage if err := json.Unmarshal(msg.Data, &m); err != nil { + sentry.CaptureException(err) logrus.WithError(err).Errorf("failed to read device message from key change topic") return true } @@ -105,6 +108,7 @@ func (t *KeyChangeConsumer) onDeviceKeyMessage(m api.DeviceMessage) bool { // only send key change events which originated from us _, originServerName, err := gomatrixserverlib.SplitID('@', m.UserID) if err != nil { + sentry.CaptureException(err) logger.WithError(err).Error("Failed to extract domain from key change event") return true } @@ -118,6 +122,7 @@ func (t *KeyChangeConsumer) onDeviceKeyMessage(m api.DeviceMessage) bool { WantMembership: "join", }, &queryRes) if err != nil { + sentry.CaptureException(err) logger.WithError(err).Error("failed to calculate joined rooms for user") return true } @@ -125,6 +130,7 @@ func (t *KeyChangeConsumer) onDeviceKeyMessage(m api.DeviceMessage) bool { // send this key change to all servers who share rooms with this user. destinations, err := t.db.GetJoinedHostsForRooms(t.ctx, queryRes.RoomIDs, true) if err != nil { + sentry.CaptureException(err) logger.WithError(err).Error("failed to calculate joined hosts for rooms user is in") return true } @@ -147,6 +153,7 @@ func (t *KeyChangeConsumer) onDeviceKeyMessage(m api.DeviceMessage) bool { Keys: m.KeyJSON, } if edu.Content, err = json.Marshal(event); err != nil { + sentry.CaptureException(err) logger.WithError(err).Error("failed to marshal EDU JSON") return true } @@ -160,6 +167,7 @@ func (t *KeyChangeConsumer) onCrossSigningMessage(m api.DeviceMessage) bool { output := m.CrossSigningKeyUpdate _, host, err := gomatrixserverlib.SplitID('@', output.UserID) if err != nil { + sentry.CaptureException(err) logrus.WithError(err).Errorf("fedsender key change consumer: user ID parse failure") return true } @@ -176,12 +184,14 @@ func (t *KeyChangeConsumer) onCrossSigningMessage(m api.DeviceMessage) bool { WantMembership: "join", }, &queryRes) if err != nil { + sentry.CaptureException(err) logger.WithError(err).Error("fedsender key change consumer: failed to calculate joined rooms for user") return true } // send this key change to all servers who share rooms with this user. destinations, err := t.db.GetJoinedHostsForRooms(t.ctx, queryRes.RoomIDs, true) if err != nil { + sentry.CaptureException(err) logger.WithError(err).Error("fedsender key change consumer: failed to calculate joined hosts for rooms user is in") return true } @@ -196,6 +206,7 @@ func (t *KeyChangeConsumer) onCrossSigningMessage(m api.DeviceMessage) bool { Origin: string(t.serverName), } if edu.Content, err = json.Marshal(output); err != nil { + sentry.CaptureException(err) logger.WithError(err).Error("fedsender key change consumer: failed to marshal output, dropping") return true } diff --git a/federationapi/consumers/sendtodevice.go b/federationapi/consumers/sendtodevice.go index ffc1d889..9aec22a3 100644 --- a/federationapi/consumers/sendtodevice.go +++ b/federationapi/consumers/sendtodevice.go @@ -18,16 +18,18 @@ import ( "context" "encoding/json" + "github.com/getsentry/sentry-go" + "github.com/matrix-org/gomatrixserverlib" + "github.com/matrix-org/util" + "github.com/nats-io/nats.go" + log "github.com/sirupsen/logrus" + "github.com/matrix-org/dendrite/federationapi/queue" "github.com/matrix-org/dendrite/federationapi/storage" "github.com/matrix-org/dendrite/setup/config" "github.com/matrix-org/dendrite/setup/jetstream" "github.com/matrix-org/dendrite/setup/process" syncTypes "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" ) // OutputSendToDeviceConsumer consumes events that originate in the clientapi. @@ -76,6 +78,7 @@ func (t *OutputSendToDeviceConsumer) onMessage(ctx context.Context, msgs []*nats sender := msg.Header.Get("sender") _, originServerName, err := gomatrixserverlib.SplitID('@', sender) if err != nil { + sentry.CaptureException(err) log.WithError(err).WithField("user_id", sender).Error("Failed to extract domain from send-to-device sender") return true } @@ -85,12 +88,14 @@ func (t *OutputSendToDeviceConsumer) onMessage(ctx context.Context, msgs []*nats // Extract the send-to-device event from msg. var ote syncTypes.OutputSendToDeviceEvent if err = json.Unmarshal(msg.Data, &ote); err != nil { + sentry.CaptureException(err) log.WithError(err).Errorf("output log: message parse failed (expected send-to-device)") return true } _, destServerName, err := gomatrixserverlib.SplitID('@', ote.UserID) if err != nil { + sentry.CaptureException(err) log.WithError(err).WithField("user_id", ote.UserID).Error("Failed to extract domain from send-to-device destination") return true } @@ -116,6 +121,7 @@ func (t *OutputSendToDeviceConsumer) onMessage(ctx context.Context, msgs []*nats }, } if edu.Content, err = json.Marshal(tdm); err != nil { + sentry.CaptureException(err) log.WithError(err).Error("failed to marshal EDU JSON") return true } diff --git a/federationapi/queue/queue.go b/federationapi/queue/queue.go index 88664fcf..8245aa5b 100644 --- a/federationapi/queue/queue.go +++ b/federationapi/queue/queue.go @@ -21,6 +21,7 @@ import ( "sync" "time" + "github.com/getsentry/sentry-go" "github.com/matrix-org/gomatrixserverlib" "github.com/prometheus/client_golang/prometheus" log "github.com/sirupsen/logrus" @@ -307,11 +308,13 @@ func (oqs *OutgoingQueues) SendEDU( ephemeralJSON, err := json.Marshal(e) if err != nil { + sentry.CaptureException(err) return fmt.Errorf("json.Marshal: %w", err) } nid, err := oqs.db.StoreJSON(oqs.process.Context(), string(ephemeralJSON)) if err != nil { + sentry.CaptureException(err) return fmt.Errorf("sendevent: oqs.db.StoreJSON: %w", err) } diff --git a/federationapi/routing/send.go b/federationapi/routing/send.go index 060af676..b3bbaa39 100644 --- a/federationapi/routing/send.go +++ b/federationapi/routing/send.go @@ -22,6 +22,7 @@ import ( "sync" "time" + "github.com/getsentry/sentry-go" "github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/util" "github.com/prometheus/client_golang/prometheus" @@ -350,6 +351,7 @@ func (t *txnReq) processEDUs(ctx context.Context) { for deviceID, message := range byUser { // TODO: check that the user and the device actually exist here if err := t.producer.SendToDevice(ctx, directPayload.Sender, userID, deviceID, directPayload.Type, message); err != nil { + sentry.CaptureException(err) util.GetLogger(ctx).WithError(err).WithFields(logrus.Fields{ "sender": directPayload.Sender, "user_id": userID, @@ -360,6 +362,7 @@ func (t *txnReq) processEDUs(ctx context.Context) { } case gomatrixserverlib.MDeviceListUpdate: if err := t.producer.SendDeviceListUpdate(ctx, e.Content, t.Origin); err != nil { + sentry.CaptureException(err) util.GetLogger(ctx).WithError(err).Error("failed to InputDeviceListUpdate") } case gomatrixserverlib.MReceipt: @@ -395,6 +398,7 @@ func (t *txnReq) processEDUs(ctx context.Context) { } case types.MSigningKeyUpdate: if err := t.producer.SendSigningKeyUpdate(ctx, e.Content, t.Origin); err != nil { + sentry.CaptureException(err) logrus.WithError(err).Errorf("Failed to process signing key update") } case gomatrixserverlib.MPresence: