Implement "full_state" query parameter for /sync (#751)

Closes #637.

syncapi/storage/syncserver.go

@@ -241,6 +241,7 @@ func (d *SyncServerDatasource) addPDUDeltaToResponse(
 	device authtypes.Device,
 	fromPos, toPos int64,
 	numRecentEventsPerRoom int,
+	wantFullState bool,
 	res *types.Response,
 ) ([]string, error) {
 	txn, err := d.db.BeginTx(ctx, &txReadOnlySnapshot)
@@ -254,7 +255,13 @@ func (d *SyncServerDatasource) addPDUDeltaToResponse(
 	// joined rooms, but also which rooms have membership transitions for this user between the 2 PDU stream positions.
 	// This works out what the 'state' key should be for each room as well as which membership block
 	// to put the room into.
-	deltas, joinedRoomIDs, err := d.getStateDeltas(ctx, &device, txn, fromPos, toPos, device.UserID)
+	var deltas []stateDelta
+	var joinedRoomIDs []string
+	if !wantFullState {
+		deltas, joinedRoomIDs, err = d.getStateDeltas(ctx, &device, txn, fromPos, toPos, device.UserID)
+	} else {
+		deltas, joinedRoomIDs, err = d.getStateDeltasForFullStateSync(ctx, &device, txn, fromPos, toPos, device.UserID)
+	}
 	if err != nil {
 		return nil, err
 	}
@@ -336,15 +343,16 @@ func (d *SyncServerDatasource) IncrementalSync(
 	device authtypes.Device,
 	fromPos, toPos types.SyncPosition,
 	numRecentEventsPerRoom int,
+	wantFullState bool,
 ) (*types.Response, error) {
 	nextBatchPos := fromPos.WithUpdates(toPos)
 	res := types.NewResponse(nextBatchPos)
 
 	var joinedRoomIDs []string
 	var err error
-	if fromPos.PDUPosition != toPos.PDUPosition {
+	if fromPos.PDUPosition != toPos.PDUPosition || wantFullState {
 		joinedRoomIDs, err = d.addPDUDeltaToResponse(
-			ctx, device, fromPos.PDUPosition, toPos.PDUPosition, numRecentEventsPerRoom, res,
+			ctx, device, fromPos.PDUPosition, toPos.PDUPosition, numRecentEventsPerRoom, wantFullState, res,
 		)
 	} else {
 		joinedRoomIDs, err = d.roomstate.selectRoomIDsWithMembership(
@@ -593,21 +601,30 @@ func (d *SyncServerDatasource) addRoomDeltaToResponse(
 	recentEvents := streamEventsToEvents(device, recentStreamEvents)
 	delta.stateEvents = removeDuplicates(delta.stateEvents, recentEvents) // roll back
 
-	// Don't bother appending empty room entries
-	if len(recentEvents) == 0 && len(delta.stateEvents) == 0 {
-		return nil
+	var prevPDUPos int64
+
+	if len(recentEvents) == 0 {
+		if len(delta.stateEvents) == 0 {
+			// Don't bother appending empty room entries
+			return nil
+		}
+
+		// If full_state=true and since is already up to date, then we'll have
+		// state events but no recent events.
+		prevPDUPos = toPos - 1
+	} else {
+		prevPDUPos = recentStreamEvents[0].streamPosition - 1
+	}
+
+	if prevPDUPos <= 0 {
+		prevPDUPos = 1
 	}
 
 	switch delta.membership {
 	case membershipJoin:
 		jr := types.NewJoinResponse()
-		if prevPDUPos := recentStreamEvents[0].streamPosition - 1; prevPDUPos > 0 {
-			// Use the short form of batch token for prev_batch
-			jr.Timeline.PrevBatch = strconv.FormatInt(prevPDUPos, 10)
-		} else {
-			// Use the short form of batch token for prev_batch
-			jr.Timeline.PrevBatch = "1"
-		}
+		// Use the short form of batch token for prev_batch
+		jr.Timeline.PrevBatch = strconv.FormatInt(prevPDUPos, 10)
 		jr.Timeline.Events = gomatrixserverlib.ToClientEvents(recentEvents, gomatrixserverlib.FormatSync)
 		jr.Timeline.Limited = false // TODO: if len(events) >= numRecents + 1 and then set limited:true
 		jr.State.Events = gomatrixserverlib.ToClientEvents(delta.stateEvents, gomatrixserverlib.FormatSync)
@@ -618,13 +635,8 @@ func (d *SyncServerDatasource) addRoomDeltaToResponse(
 		// TODO: recentEvents may contain events that this user is not allowed to see because they are
 		//       no longer in the room.
 		lr := types.NewLeaveResponse()
-		if prevPDUPos := recentStreamEvents[0].streamPosition - 1; prevPDUPos > 0 {
-			// Use the short form of batch token for prev_batch
-			lr.Timeline.PrevBatch = strconv.FormatInt(prevPDUPos, 10)
-		} else {
-			// Use the short form of batch token for prev_batch
-			lr.Timeline.PrevBatch = "1"
-		}
+		// Use the short form of batch token for prev_batch
+		lr.Timeline.PrevBatch = strconv.FormatInt(prevPDUPos, 10)
 		lr.Timeline.Events = gomatrixserverlib.ToClientEvents(recentEvents, gomatrixserverlib.FormatSync)
 		lr.Timeline.Limited = false // TODO: if len(events) >= numRecents + 1 and then set limited:true
 		lr.State.Events = gomatrixserverlib.ToClientEvents(delta.stateEvents, gomatrixserverlib.FormatSync)
@@ -758,15 +770,11 @@ func (d *SyncServerDatasource) getStateDeltas(
 			if membership := getMembershipFromEvent(&ev.Event, userID); membership != "" {
 				if membership == membershipJoin {
 					// send full room state down instead of a delta
-					var allState []gomatrixserverlib.Event
-					allState, err = d.roomstate.selectCurrentState(ctx, txn, roomID)
+					var s []streamEvent
+					s, err = d.currentStateStreamEventsForRoom(ctx, txn, roomID)
 					if err != nil {
 						return nil, nil, err
 					}
-					s := make([]streamEvent, len(allState))
-					for i := 0; i < len(s); i++ {
-						s[i] = streamEvent{Event: allState[i], streamPosition: 0}
-					}
 					state[roomID] = s
 					continue // we'll add this room in when we do joined rooms
 				}
@@ -798,6 +806,79 @@ func (d *SyncServerDatasource) getStateDeltas(
 	return deltas, joinedRoomIDs, nil
 }
 
+// getStateDeltasForFullStateSync is a variant of getStateDeltas used for /sync
+// requests with full_state=true.
+// Fetches full state for all joined rooms and uses selectStateInRange to get
+// updates for other rooms.
+func (d *SyncServerDatasource) getStateDeltasForFullStateSync(
+	ctx context.Context, device *authtypes.Device, txn *sql.Tx,
+	fromPos, toPos int64, userID string,
+) ([]stateDelta, []string, error) {
+	joinedRoomIDs, err := d.roomstate.selectRoomIDsWithMembership(ctx, txn, userID, "join")
+	if err != nil {
+		return nil, nil, err
+	}
+
+	// Use a reasonable initial capacity
+	deltas := make([]stateDelta, 0, len(joinedRoomIDs))
+
+	// Add full states for all joined rooms
+	for _, joinedRoomID := range joinedRoomIDs {
+		s, stateErr := d.currentStateStreamEventsForRoom(ctx, txn, joinedRoomID)
+		if stateErr != nil {
+			return nil, nil, stateErr
+		}
+		deltas = append(deltas, stateDelta{
+			membership:  "join",
+			stateEvents: streamEventsToEvents(device, s),
+			roomID:      joinedRoomID,
+		})
+	}
+
+	// Get all the state events ever between these two positions
+	stateNeeded, eventMap, err := d.events.selectStateInRange(ctx, txn, fromPos, toPos)
+	if err != nil {
+		return nil, nil, err
+	}
+	state, err := d.fetchStateEvents(ctx, txn, stateNeeded, eventMap)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	for roomID, stateStreamEvents := range state {
+		for _, ev := range stateStreamEvents {
+			if membership := getMembershipFromEvent(&ev.Event, userID); membership != "" {
+				if membership != "join" { // We've already added full state for all joined rooms above.
+					deltas = append(deltas, stateDelta{
+						membership:    membership,
+						membershipPos: ev.streamPosition,
+						stateEvents:   streamEventsToEvents(device, stateStreamEvents),
+						roomID:        roomID,
+					})
+				}
+
+				break
+			}
+		}
+	}
+
+	return deltas, joinedRoomIDs, nil
+}
+
+func (d *SyncServerDatasource) currentStateStreamEventsForRoom(
+	ctx context.Context, txn *sql.Tx, roomID string,
+) ([]streamEvent, error) {
+	allState, err := d.roomstate.selectCurrentState(ctx, txn, roomID)
+	if err != nil {
+		return nil, err
+	}
+	s := make([]streamEvent, len(allState))
+	for i := 0; i < len(s); i++ {
+		s[i] = streamEvent{Event: allState[i], streamPosition: 0}
+	}
+	return s, nil
+}
+
 // streamEventsToEvents converts streamEvent to Event. If device is non-nil and
 // matches the streamevent.transactionID device then the transaction ID gets
 // added to the unsigned section of the output event.

syncapi/sync/requestpool.go

@@ -65,8 +65,7 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *authtype
 
 	currPos := rp.notifier.CurrentPosition()
 
-	// If this is an initial sync or timeout=0 we return immediately
-	if syncReq.since == nil || syncReq.timeout == 0 {
+	if shouldReturnImmediately(syncReq) {
 		syncData, err = rp.currentSyncForUser(*syncReq, currPos)
 		if err != nil {
 			return httputil.LogThenError(req, err)
@@ -135,7 +134,7 @@ func (rp *RequestPool) currentSyncForUser(req syncRequest, latestPos types.SyncP
 	if req.since == nil {
 		res, err = rp.db.CompleteSync(req.ctx, req.device.UserID, req.limit)
 	} else {
-		res, err = rp.db.IncrementalSync(req.ctx, req.device, *req.since, latestPos, req.limit)
+		res, err = rp.db.IncrementalSync(req.ctx, req.device, *req.since, latestPos, req.limit, req.wantFullState)
 	}
 
 	if err != nil {
@@ -216,3 +215,10 @@ func (rp *RequestPool) appendAccountData(
 
 	return data, nil
 }
+
+// shouldReturnImmediately returns whether the /sync request is an initial sync,
+// or timeout=0, or full_state=true, in any of the cases the request should
+// return immediately.
+func shouldReturnImmediately(syncReq *syncRequest) bool {
+	return syncReq.since == nil || syncReq.timeout == 0 || syncReq.wantFullState
+}

testfile

@@ -158,3 +158,4 @@ Inbound federation rejects remote attempts to join local users to rooms
 Inbound federation rejects remote attempts to kick local users to rooms
 An event which redacts itself should be ignored
 A pair of events which redact each other should be ignored
+Full state sync includes joined rooms