aboutsummaryrefslogtreecommitdiff
path: root/syncapi
diff options
context:
space:
mode:
authorNeil Alexander <neilalexander@users.noreply.github.com>2020-10-02 12:50:58 +0100
committerGitHub <noreply@github.com>2020-10-02 12:50:58 +0100
commitfb9a8f215b0af25c02d62c6f44ed91adc3349a6e (patch)
tree0a0c1ccea79d2acb3f46f43223e1293f178a56df /syncapi
parent1b29e5771feb335afe0e39e7f33f1d1896dbdb2f (diff)
Fix initial sync (#1465)
* Fix complete sync check * Remove unnecessary 'since' copy * Fix failing test * Un-whitelist a couple of tests Co-authored-by: Kegan Dougal <kegan@matrix.org>
Diffstat (limited to 'syncapi')
-rw-r--r--syncapi/storage/shared/syncserver.go17
-rw-r--r--syncapi/sync/requestpool.go13
2 files changed, 13 insertions, 17 deletions
diff --git a/syncapi/storage/shared/syncserver.go b/syncapi/storage/shared/syncserver.go
index 05a8768e..edb51b34 100644
--- a/syncapi/storage/shared/syncserver.go
+++ b/syncapi/storage/shared/syncserver.go
@@ -670,7 +670,7 @@ func (d *Database) RedactEvent(ctx context.Context, redactedEventID string, reda
// nolint:nakedret
func (d *Database) getResponseWithPDUsForCompleteSync(
ctx context.Context, res *types.Response,
- userID string, deviceID string,
+ userID string, device userapi.Device,
numRecentEventsPerRoom int,
) (
toPos types.StreamingToken,
@@ -712,7 +712,7 @@ func (d *Database) getResponseWithPDUsForCompleteSync(
for _, roomID := range joinedRoomIDs {
var jr *types.JoinResponse
jr, err = d.getJoinResponseForCompleteSync(
- ctx, txn, roomID, r, &stateFilter, numRecentEventsPerRoom,
+ ctx, txn, roomID, r, &stateFilter, numRecentEventsPerRoom, device,
)
if err != nil {
return
@@ -721,7 +721,7 @@ func (d *Database) getResponseWithPDUsForCompleteSync(
}
// Add peeked rooms.
- peeks, err := d.Peeks.SelectPeeksInRange(ctx, txn, userID, deviceID, r)
+ peeks, err := d.Peeks.SelectPeeksInRange(ctx, txn, userID, device.ID, r)
if err != nil {
return
}
@@ -729,7 +729,7 @@ func (d *Database) getResponseWithPDUsForCompleteSync(
if !peek.Deleted {
var jr *types.JoinResponse
jr, err = d.getJoinResponseForCompleteSync(
- ctx, txn, peek.RoomID, r, &stateFilter, numRecentEventsPerRoom,
+ ctx, txn, peek.RoomID, r, &stateFilter, numRecentEventsPerRoom, device,
)
if err != nil {
return
@@ -751,7 +751,7 @@ func (d *Database) getJoinResponseForCompleteSync(
roomID string,
r types.Range,
stateFilter *gomatrixserverlib.StateFilter,
- numRecentEventsPerRoom int,
+ numRecentEventsPerRoom int, device userapi.Device,
) (jr *types.JoinResponse, err error) {
var stateEvents []gomatrixserverlib.HeaderedEvent
stateEvents, err = d.CurrentRoomState.SelectCurrentState(ctx, txn, roomID, stateFilter)
@@ -784,8 +784,9 @@ func (d *Database) getJoinResponseForCompleteSync(
}
// We don't include a device here as we don't need to send down
- // transaction IDs for complete syncs
- recentEvents := d.StreamEventsToEvents(nil, recentStreamEvents)
+ // transaction IDs for complete syncs, but we do it anyway because Sytest demands it for:
+ // "Can sync a room with a message with a transaction id" - which does a complete sync to check.
+ recentEvents := d.StreamEventsToEvents(&device, recentStreamEvents)
stateEvents = removeDuplicates(stateEvents, recentEvents)
jr = types.NewJoinResponse()
jr.Timeline.PrevBatch = prevBatchStr
@@ -800,7 +801,7 @@ func (d *Database) CompleteSync(
device userapi.Device, numRecentEventsPerRoom int,
) (*types.Response, error) {
toPos, joinedRoomIDs, err := d.getResponseWithPDUsForCompleteSync(
- ctx, res, device.UserID, device.ID, numRecentEventsPerRoom,
+ ctx, res, device.UserID, device, numRecentEventsPerRoom,
)
if err != nil {
return nil, fmt.Errorf("d.getResponseWithPDUsForCompleteSync: %w", err)
diff --git a/syncapi/sync/requestpool.go b/syncapi/sync/requestpool.go
index aaaf9491..8a79737a 100644
--- a/syncapi/sync/requestpool.go
+++ b/syncapi/sync/requestpool.go
@@ -197,19 +197,14 @@ func (rp *RequestPool) OnIncomingKeyChangeRequest(req *http.Request, device *use
func (rp *RequestPool) currentSyncForUser(req syncRequest, latestPos types.StreamingToken) (*types.Response, error) {
res := types.NewResponse()
- since := types.NewStreamToken(0, 0, nil)
- if req.since != nil {
- since = *req.since
- }
-
// See if we have any new tasks to do for the send-to-device messaging.
- events, updates, deletions, err := rp.db.SendToDeviceUpdatesForSync(req.ctx, req.device.UserID, req.device.ID, since)
+ events, updates, deletions, err := rp.db.SendToDeviceUpdatesForSync(req.ctx, req.device.UserID, req.device.ID, *req.since)
if err != nil {
return nil, fmt.Errorf("rp.db.SendToDeviceUpdatesForSync: %w", err)
}
// TODO: handle ignored users
- if req.since == nil {
+ if req.since.PDUPosition() == 0 && req.since.EDUPosition() == 0 {
res, err = rp.db.CompleteSync(req.ctx, res, req.device, req.limit)
if err != nil {
return res, fmt.Errorf("rp.db.CompleteSync: %w", err)
@@ -226,7 +221,7 @@ func (rp *RequestPool) currentSyncForUser(req syncRequest, latestPos types.Strea
if err != nil {
return res, fmt.Errorf("rp.appendAccountData: %w", err)
}
- res, err = rp.appendDeviceLists(res, req.device.UserID, since, latestPos)
+ res, err = rp.appendDeviceLists(res, req.device.UserID, *req.since, latestPos)
if err != nil {
return res, fmt.Errorf("rp.appendDeviceLists: %w", err)
}
@@ -240,7 +235,7 @@ func (rp *RequestPool) currentSyncForUser(req syncRequest, latestPos types.Strea
// Then add the updates into the sync response.
if len(updates) > 0 || len(deletions) > 0 {
// Handle the updates and deletions in the database.
- err = rp.db.CleanSendToDeviceUpdates(context.Background(), updates, deletions, since)
+ err = rp.db.CleanSendToDeviceUpdates(context.Background(), updates, deletions, *req.since)
if err != nil {
return res, fmt.Errorf("rp.db.CleanSendToDeviceUpdates: %w", err)
}