Some review comment fixes

federationapi/routing/send.go

@@ -515,7 +515,7 @@ func (t *txnReq) processEDUs(ctx context.Context) {
 			}
 			inputRes := &eduserverAPI.InputCrossSigningKeyUpdateResponse{}
 			if err := t.eduAPI.InputCrossSigningKeyUpdate(ctx, inputReq, inputRes); err != nil {
-				util.GetLogger(ctx).WithError(err).Error("Failed to send signing key update to EDU server")
+				util.GetLogger(ctx).WithError(err).Error("Failed to unmarshal cross-signing update")
 				continue
 			}
 		default:

federationsender/consumers/keychange.go

@@ -150,7 +150,7 @@ func (t *KeyChangeConsumer) onCrossSigningMessage(m api.DeviceMessage) error {
 	output := m.CrossSigningKeyUpdate
 	_, host, err := gomatrixserverlib.SplitID('@', output.UserID)
 	if err != nil {
-		logrus.WithError(err).Errorf("eduserver output log: user ID parse failure")
+		logrus.WithError(err).Errorf("fedsender key change consumer: user ID parse failure")
 		return nil
 	}
 	if host != gomatrixserverlib.ServerName(t.serverName) {
@@ -166,13 +166,13 @@ func (t *KeyChangeConsumer) onCrossSigningMessage(m api.DeviceMessage) error {
 		WantMembership: "join",
 	}, &queryRes)
 	if err != nil {
-		logger.WithError(err).Error("failed to calculate joined rooms for user")
+		logger.WithError(err).Error("fedsender key change consumer: failed to calculate joined rooms for user")
 		return nil
 	}
 	// send this key change to all servers who share rooms with this user.
 	destinations, err := t.db.GetJoinedHostsForRooms(context.Background(), queryRes.RoomIDs)
 	if err != nil {
-		logger.WithError(err).Error("failed to calculate joined hosts for rooms user is in")
+		logger.WithError(err).Error("fedsender key change consumer: failed to calculate joined hosts for rooms user is in")
 		return nil
 	}
 
@@ -182,7 +182,8 @@ func (t *KeyChangeConsumer) onCrossSigningMessage(m api.DeviceMessage) error {
 		Origin: string(t.serverName),
 	}
 	if edu.Content, err = json.Marshal(output); err != nil {
-		return err
+		logger.WithError(err).Error("fedsender key change consumer: failed to marshal output, dropping")
+		return nil
 	}
 
 	logger.Infof("Sending cross-signing update message to %q", destinations)

keyserver/consumers/cross_signing.go

@@ -43,9 +43,13 @@ func NewOutputCrossSigningKeyUpdateConsumer(
 	keyDB storage.Database,
 	keyAPI api.KeyInternalAPI,
 ) *OutputCrossSigningKeyUpdateConsumer {
+	// The keyserver both produces and consumes on the TopicOutputKeyChangeEvent
+	// topic. We will only produce events where the UserID matches our server name,
+	// and we will only consume events where the UserID does NOT match our server
+	// name (because the update came from a remote server).
 	consumer := internal.ContinualConsumer{
 		Process:        process,
-		ComponentName:  "keyserver/crosssigning",
+		ComponentName:  "keyserver/keyserver",
 		Topic:          cfg.Global.Kafka.TopicFor(config.TopicOutputKeyChangeEvent),
 		Consumer:       kafkaConsumer,
 		PartitionStore: keyDB,

keyserver/internal/cross_signing.go

@@ -59,8 +59,8 @@ func sanityCheckKey(key gomatrixserverlib.CrossSigningKey, userID string, purpos
 			// We can't enforce the key length to be correct for an
 			// algorithm that we don't recognise, so instead we'll
 			// just make sure that it isn't incredibly excessive.
-			if len(keyData) > 4096 {
+			if l := len(keyData); l > 4096 {
-				return fmt.Errorf("unknown key type is too long")
+				return fmt.Errorf("unknown key type is too long (%d bytes)", l)
 			}
 		}
 	}
@@ -76,8 +76,8 @@ func sanityCheckKey(key gomatrixserverlib.CrossSigningKey, userID string, purpos
 			case "curve25519":
 				return fmt.Errorf("curve25519 signatures are impossible")
 			default:
-				if len(originSignature) > 4096 {
+				if l := len(originSignature); l > 4096 {
-					return fmt.Errorf("unknown signature type is too long")
+					return fmt.Errorf("unknown signature type is too long (%d bytes)", l)
 				}
 			}
 		}

sytest-whitelist

@@ -554,3 +554,5 @@ Can upload self-signing keys
 Fails to upload self-signing keys with no auth
 Fails to upload self-signing key without master key
 can fetch self-signing keys over federation
+Changing master key notifies local users
+Changing user-signing key notifies local users