Fix positions, add ApplyUpdates

syncapi/internal/keychange.go

@@ -134,7 +134,7 @@ func DeviceListCatchup(
 		Partition: queryRes.Partition,
 		Offset:    queryRes.Offset,
 	})
-	res.NextBatch = res.NextBatch.WithUpdates(to)
+	res.NextBatch.ApplyUpdates(to)
 
 	return hasNew, nil
 }

syncapi/storage/shared/syncserver.go

@@ -575,6 +575,7 @@ func (d *Database) addTypingDeltaToResponse(
 				jr = *types.NewJoinResponse()
 			}
 			jr.Ephemeral.Events = append(jr.Ephemeral.Events, ev)
+			res.NextBatch.TypingPosition++
 			res.Rooms.Join[roomID] = jr
 		}
 	}
@@ -630,6 +631,7 @@ func (d *Database) addReceiptDeltaToResponse(
 		}
 
 		jr.Ephemeral.Events = append(jr.Ephemeral.Events, ev)
+		res.NextBatch.ReceiptPosition++
 		res.Rooms.Join[roomID] = jr
 	}
 
@@ -686,7 +688,7 @@ func (d *Database) IncrementalSync(
 	numRecentEventsPerRoom int,
 	wantFullState bool,
 ) (*types.Response, error) {
-	res.NextBatch = res.NextBatch.WithUpdates(toPos)
+	res.NextBatch = fromPos.WithUpdates(toPos)
 
 	var joinedRoomIDs []string
 	var err error
@@ -778,7 +780,7 @@ func (d *Database) getResponseWithPDUsForCompleteSync(
 		To:   toPos.PDUPosition,
 	}
 
-	res.NextBatch = res.NextBatch.WithUpdates(toPos)
+	res.NextBatch.ApplyUpdates(toPos)
 
 	// Extract room state and recent events for all rooms the user is joined to.
 	joinedRoomIDs, err = d.CurrentRoomState.SelectRoomIDsWithMembership(ctx, txn, userID, gomatrixserverlib.Join)

syncapi/sync/requestpool.go

@@ -427,7 +427,7 @@ func (rp *RequestPool) appendAccountData(
 // or timeout=0, or full_state=true, in any of the cases the request should
 // return immediately.
 func (rp *RequestPool) shouldReturnImmediately(syncReq *syncRequest) bool {
-	if syncReq.since == nil || syncReq.timeout == 0 || syncReq.wantFullState {
+	if syncReq.since.IsEmpty() || syncReq.timeout == 0 || syncReq.wantFullState {
 		return true
 	}
 	waiting, werr := rp.db.SendToDeviceUpdatesWaiting(context.TODO(), syncReq.device.UserID, syncReq.device.ID)

syncapi/types/types.go

@@ -187,28 +187,33 @@ func (t *StreamingToken) IsEmpty() bool {
 // If the latter StreamingToken contains a field that is not 0, it is considered an update,
 // and its value will replace the corresponding value in the StreamingToken on which WithUpdates is called.
 // If the other token has a log, they will replace any existing log on this token.
-func (t *StreamingToken) WithUpdates(other StreamingToken) (ret StreamingToken) {
-	ret = *t
+func (t *StreamingToken) WithUpdates(other StreamingToken) StreamingToken {
+	ret := *t
+	ret.ApplyUpdates(other)
+	return ret
+}
+
+// ApplyUpdates applies any changes from the supplied StreamingToken. If the supplied
+// streaming token contains any positions that are not 0, they are considered updates
+// and will overwrite the value in the token.
+func (t *StreamingToken) ApplyUpdates(other StreamingToken) {
 	switch {
 	case other.PDUPosition > 0:
-		ret.PDUPosition = other.PDUPosition
+		t.PDUPosition = other.PDUPosition
 	case other.TypingPosition > 0:
-		ret.TypingPosition = other.TypingPosition
+		t.TypingPosition = other.TypingPosition
 	case other.ReceiptPosition > 0:
-		ret.ReceiptPosition = other.ReceiptPosition
+		t.ReceiptPosition = other.ReceiptPosition
 	case other.SendToDevicePosition > 0:
-		ret.SendToDevicePosition = other.SendToDevicePosition
+		t.SendToDevicePosition = other.SendToDevicePosition
 	}
-	ret.Logs = make(map[string]*LogPosition)
-	for name := range t.Logs {
-		otherLog := other.Log(name)
-		if otherLog == nil {
-			continue
-		}
-		copy := *otherLog
-		ret.Logs[name] = &copy
+	if t.Logs == nil {
+		t.Logs = make(map[string]*LogPosition)
+	}
+	for name, pos := range other.Logs {
+		copy := *pos
+		t.Logs[name] = &copy
 	}
-	return ret
 }
 
 type TopologyToken struct {