Peeking over federation via MSC2444

docs/peeking.md

@@ -1,19 +1,26 @@
 ## Peeking
 
-Peeking is implemented as per [MSC2753](https://github.com/matrix-org/matrix-doc/pull/2753).
+Local peeking is implemented as per [MSC2753](https://github.com/matrix-org/matrix-doc/pull/2753).
 
 Implementationwise, this means:
  * Users call `/peek` and `/unpeek` on the clientapi from a given device.
  * The clientapi delegates these via HTTP to the roomserver, which coordinates peeking in general for a given room
  * The roomserver writes an NewPeek event into the kafka log headed to the syncserver
- * The syncserver tracks the existence of the local peek in its DB, and then starts waking up the peeking devices for the room in question, putting it in the `peek` section of the /sync response.
+ * The syncserver tracks the existence of the local peek in the syncapi_peeks table in its DB, and then starts waking up the peeking devices for the room in question, putting it in the `peek` section of the /sync response.
 
-Questions (given this is [my](https://github.com/ara4n) first time hacking on Dendrite):
- * The whole clientapi -> roomserver -> syncapi flow to initiate a peek seems very indirect.  Is there a reason not to just let syncapi itself host the implementation of `/peek`?
+Peeking over federation is implemented as per [MSC2444](https://github.com/matrix-org/matrix-doc/pull/2444).
 
-In future, peeking over federation will be added as per [MSC2444](https://github.com/matrix-org/matrix-doc/pull/2444).
- * The `roomserver` will kick the `federationsender` much as it does for a federated `/join` in order to trigger a federated `/peek`
- * The `federationsender` tracks the existence of the remote peek in question
+For requests to peek our rooms ("inbound peeks"):
+ * Remote servers call `/peek` on federationapi
+   * The federationapi queries the federationsender to check if this is renewing an inbound peek or not.
+   * If not, it hits the PerformInboundPeek on the roomserver to ask it for the current state of the room.
+   * The roomserver atomically (in theory) adds a NewInboundPeek to its kafka stream to tell the federationserver to start peeking.
+   * The federationsender receives the event, tracks the inbound peek in the federationsender_inbound_peeks table, and starts sending events to the peeking server.
+   * The federationsender evicts stale inbound peeks which haven't been renewed.
+
+For peeking into other server's rooms ("outbound peeks"):
+ * The `roomserver` will kick the `federationsender` much as it does for a federated `/join` in order to trigger a federated outbound `/peek`
+ * The `federationsender` tracks the existence of the outbound peek in in its federationsender_outbound_peeks table.
  * The `federationsender` regularly renews the remote peek as long as there are still peeking devices syncing for it.
   * TBD: how do we tell if there are no devices currently syncing for a given peeked room?  The syncserver needs to tell the roomserver
     somehow who then needs to warn the federationsender.

federationapi/routing/peek.go

@@ -0,0 +1,102 @@
+// Copyright 2020 New Vector Ltd
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package routing
+
+import (
+	"net/http"
+
+	"github.com/matrix-org/dendrite/clientapi/jsonerror"
+	"github.com/matrix-org/dendrite/roomserver/api"
+	"github.com/matrix-org/dendrite/setup/config"
+	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/matrix-org/util"
+)
+
+// Peek implements the SS /peek API, handling inbound peeks
+func Peek(
+	httpReq *http.Request,
+	request *gomatrixserverlib.FederationRequest,
+	cfg *config.FederationAPI,
+	rsAPI api.RoomserverInternalAPI,
+	roomID, peekID string,
+	remoteVersions []gomatrixserverlib.RoomVersion,
+) util.JSONResponse {
+	// TODO: check if we're just refreshing an existing peek by querying the federationsender
+
+	verReq := api.QueryRoomVersionForRoomRequest{RoomID: roomID}
+	verRes := api.QueryRoomVersionForRoomResponse{}
+	if err := rsAPI.QueryRoomVersionForRoom(httpReq.Context(), &verReq, &verRes); err != nil {
+		return util.JSONResponse{
+			Code: http.StatusInternalServerError,
+			JSON: jsonerror.InternalServerError(),
+		}
+	}
+
+	// Check that the room that the peeking server is trying to peek is actually
+	// one of the room versions that they listed in their supported ?ver= in
+	// the peek URL.
+	remoteSupportsVersion := false
+	for _, v := range remoteVersions {
+		if v == verRes.RoomVersion {
+			remoteSupportsVersion = true
+			break
+		}
+	}
+	// If it isn't, stop trying to peek the room.
+	if !remoteSupportsVersion {
+		return util.JSONResponse{
+			Code: http.StatusBadRequest,
+			JSON: jsonerror.IncompatibleRoomVersion(verRes.RoomVersion),
+		}
+	}
+
+	// TODO: Check history visibility
+
+	// tell the peeking server to renew every hour
+	renewalInterval := int64(60 * 60 * 1000 * 1000)
+
+	var response api.PerformInboundPeekResponse
+	err := rsAPI.PerformInboundPeek(
+		httpReq.Context(),
+		&api.PerformInboundPeekRequest{
+			RoomID:          roomID,
+			PeekID:          peekID,
+			ServerName:      request.Origin(),
+			RenewalInterval: renewalInterval,
+		},
+		&response,
+	)
+	if err != nil {
+		resErr := util.ErrorResponse(err)
+		return resErr
+	}
+
+	if !response.RoomExists {
+		return util.JSONResponse{Code: http.StatusNotFound, JSON: nil}
+	}
+
+	respPeek := gomatrixserverlib.RespPeek{
+		StateEvents:     gomatrixserverlib.UnwrapEventHeaders(response.StateEvents),
+		AuthEvents:      gomatrixserverlib.UnwrapEventHeaders(response.AuthChainEvents),
+		RoomVersion:     response.RoomVersion,
+		LatestEvent:     response.LatestEvent.Unwrap(),
+		RenewalInterval: renewalInterval,
+	}
+
+	return util.JSONResponse{
+		Code: http.StatusOK,
+		JSON: respPeek,
+	}
+}

federationapi/routing/routing.go

@@ -229,7 +229,37 @@ func Setup(
 		},
 	)).Methods(http.MethodGet)
 
-	v1fedmux.Handle("/make_join/{roomID}/{eventID}", httputil.MakeFedAPI(
+	v1fedmux.Handle("/peek/{roomID}/{peekID}", httputil.MakeFedAPI(
+		"federation_peek", cfg.Matrix.ServerName, keys, wakeup,
+		func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
+			if roomserverAPI.IsServerBannedFromRoom(httpReq.Context(), rsAPI, vars["roomID"], request.Origin()) {
+				return util.JSONResponse{
+					Code: http.StatusForbidden,
+					JSON: jsonerror.Forbidden("Forbidden by server ACLs"),
+				}
+			}
+			roomID := vars["roomID"]
+			peekID := vars["peekID"]
+			queryVars := httpReq.URL.Query()
+			remoteVersions := []gomatrixserverlib.RoomVersion{}
+			if vers, ok := queryVars["ver"]; ok {
+				// The remote side supplied a ?ver= so use that to build up the list
+				// of supported room versions
+				for _, v := range vers {
+					remoteVersions = append(remoteVersions, gomatrixserverlib.RoomVersion(v))
+				}
+			} else {
+				// The remote side didn't supply a ?ver= so just assume that they only
+				// support room version 1
+				remoteVersions = append(remoteVersions, gomatrixserverlib.RoomVersionV1)
+			}
+			return Peek(
+				httpReq, request, cfg, rsAPI, roomID, peekID, remoteVersions,
+			)
+		},
+	)).Methods(http.MethodPut, http.MethodDelete)
+
+	v1fedmux.Handle("/make_join/{roomID}/{userID}", httputil.MakeFedAPI(
 		"federation_make_join", cfg.Matrix.ServerName, keys, wakeup,
 		func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
 			if roomserverAPI.IsServerBannedFromRoom(httpReq.Context(), rsAPI, vars["roomID"], request.Origin()) {
@@ -239,11 +269,11 @@ func Setup(
 				}
 			}
 			roomID := vars["roomID"]
-			eventID := vars["eventID"]
+			userID := vars["userID"]
 			queryVars := httpReq.URL.Query()
 			remoteVersions := []gomatrixserverlib.RoomVersion{}
 			if vers, ok := queryVars["ver"]; ok {
-				// The remote side supplied a ?=ver so use that to build up the list
+				// The remote side supplied a ?ver= so use that to build up the list
 				// of supported room versions
 				for _, v := range vers {
 					remoteVersions = append(remoteVersions, gomatrixserverlib.RoomVersion(v))
@@ -255,7 +285,7 @@ func Setup(
 				remoteVersions = append(remoteVersions, gomatrixserverlib.RoomVersionV1)
 			}
 			return MakeJoin(
-				httpReq, request, cfg, rsAPI, roomID, eventID, remoteVersions,
+				httpReq, request, cfg, rsAPI, roomID, userID, remoteVersions,
 			)
 		},
 	)).Methods(http.MethodGet)

federationsender/api/api.go

@@ -62,6 +62,12 @@ type FederationSenderInternalAPI interface {
 		request *PerformJoinRequest,
 		response *PerformJoinResponse,
 	)
+	// Handle an instruction to peek a room on a remote server.
+	PerformOutboundPeek(
+		ctx context.Context,
+		request *PerformOutboundPeekRequest,
+		response *PerformOutboundPeekResponse,
+	) error
 	// Handle an instruction to make_leave & send_leave with a remote server.
 	PerformLeave(
 		ctx context.Context,
@@ -111,6 +117,16 @@ type PerformJoinResponse struct {
 	LastError *gomatrix.HTTPError
 }
 
+type PerformOutboundPeekRequest struct {
+	RoomID string `json:"room_id"`
+	// The sorted list of servers to try. Servers will be tried sequentially, after de-duplication.
+	ServerNames types.ServerNames `json:"server_names"`
+}
+
+type PerformOutboundPeekResponse struct {
+	LastError *gomatrix.HTTPError
+}
+
 type PerformLeaveRequest struct {
 	RoomID      string            `json:"room_id"`
 	UserID      string            `json:"user_id"`

federationsender/consumers/roomserver.go

@@ -111,6 +111,14 @@ func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error {
 			}
 			return nil
 		}
+	case api.OutputTypeNewInboundPeek:
+		if err := s.processInboundPeek(*output.NewInboundPeek); err != nil {
+			log.WithFields(log.Fields{
+				"event":      output.NewInboundPeek,
+				log.ErrorKey: err,
+			}).Panicf("roomserver output log: remote peek event failure")
+			return nil
+		}
 	default:
 		log.WithField("type", output.Type).Debug(
 			"roomserver output log: ignoring unknown output type",
@@ -121,6 +129,23 @@ func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error {
 	return nil
 }
 
+// processInboundPeek starts tracking a new federated inbound peek (replacing the existing one if any)
+// causing the federationsender to start sending messages to the peeking server
+func (s *OutputRoomEventConsumer) processInboundPeek(orp api.OutputNewInboundPeek) error {
+
+	// FIXME: there's a race here - we should start /sending new peeked events
+	// atomically after the orp.LatestEventID to ensure there are no gaps between
+	// the peek beginning and the send stream beginning.
+	//
+	// We probably need to track orp.LatestEventID on the inbound peek, but it's
+	// unclear how we then use that to prevent the race when we start the send
+	// stream.
+	//
+	// This is making the tests flakey.
+
+	return s.db.AddInboundPeek(context.TODO(), orp.ServerName, orp.RoomID, orp.PeekID, orp.RenewalInterval)
+}
+
 // processMessage updates the list of currently joined hosts in the room
 // and then sends the event to the hosts that were joined before the event.
 func (s *OutputRoomEventConsumer) processMessage(ore api.OutputNewRoomEvent) error {
@@ -164,14 +189,18 @@ func (s *OutputRoomEventConsumer) processMessage(ore api.OutputNewRoomEvent) err
 		return err
 	}
 
+	// TODO: do housekeeping to evict unrenewed peeking hosts
+
+	// TODO: implement query to let the fedapi check whether a given peek is live or not
+
 	// Send the event.
 	return s.queues.SendEvent(
 		ore.Event, gomatrixserverlib.ServerName(ore.SendAsServer), joinedHostsAtEvent,
 	)
 }
 
 // joinedHostsAtEvent works out a list of matrix servers that were joined to
-// the room at the event.
+// the room at the event (including peeking ones)
 // It is important to use the state at the event for sending messages because:
 //   1) We shouldn't send messages to servers that weren't in the room.
 //   2) If a server is kicked from the rooms it should still be told about the
@@ -222,6 +251,15 @@ func (s *OutputRoomEventConsumer) joinedHostsAtEvent(
 		joined[joinedHost.ServerName] = true
 	}
 
+	// handle peeking hosts
+	inboundPeeks, err := s.db.GetInboundPeeks(context.TODO(), ore.Event.Event.RoomID())
+	if err != nil {
+		return nil, err
+	}
+	for _, inboundPeek := range inboundPeeks {
+		joined[inboundPeek.ServerName] = true
+	}
+
 	var result []gomatrixserverlib.ServerName
 	for serverName, include := range joined {
 		if include {