diff options
author | Neil Alexander <neilalexander@users.noreply.github.com> | 2020-10-02 12:50:58 +0100 |
---|---|---|
committer | GitHub <noreply@github.com> | 2020-10-02 12:50:58 +0100 |
commit | fb9a8f215b0af25c02d62c6f44ed91adc3349a6e (patch) | |
tree | 0a0c1ccea79d2acb3f46f43223e1293f178a56df /syncapi | |
parent | 1b29e5771feb335afe0e39e7f33f1d1896dbdb2f (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.go | 17 | ||||
-rw-r--r-- | syncapi/sync/requestpool.go | 13 |
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) } |