diff --git a/syncapi/consumers/eduserver_sendtodevice.go b/syncapi/consumers/eduserver_sendtodevice.go index 6e774b5b4..d95067921 100644 --- a/syncapi/consumers/eduserver_sendtodevice.go +++ b/syncapi/consumers/eduserver_sendtodevice.go @@ -105,6 +105,8 @@ func (s *OutputSendToDeviceEventConsumer) onMessage(msg *sarama.ConsumerMessage) return err } + util.GetLogger(context.TODO()).Infof("Stored send-to-device message at position %d", streamPos) + s.stream.Advance(streamPos) s.notifier.OnNewSendToDevice( output.UserID, diff --git a/syncapi/storage/postgres/send_to_device_table.go b/syncapi/storage/postgres/send_to_device_table.go index 11e00fc14..3f4208906 100644 --- a/syncapi/storage/postgres/send_to_device_table.go +++ b/syncapi/storage/postgres/send_to_device_table.go @@ -50,7 +50,7 @@ const insertSendToDeviceMessageSQL = ` const selectSendToDeviceMessagesSQL = ` SELECT id, user_id, device_id, content FROM syncapi_send_to_device - WHERE user_id = $1 AND device_id = $2 AND id > $3 AND id <= $4 + WHERE user_id = $1 AND device_id = $2 AND id <= $3 ORDER BY id DESC ` @@ -98,16 +98,16 @@ func (s *sendToDeviceStatements) InsertSendToDeviceMessage( } func (s *sendToDeviceStatements) SelectSendToDeviceMessages( - ctx context.Context, txn *sql.Tx, userID, deviceID string, from, to types.StreamPosition, + ctx context.Context, txn *sql.Tx, userID, deviceID string, to types.StreamPosition, ) (lastPos types.StreamPosition, events []types.SendToDeviceEvent, err error) { - rows, err := sqlutil.TxStmt(txn, s.selectSendToDeviceMessagesStmt).QueryContext(ctx, userID, deviceID, from, to) + rows, err := sqlutil.TxStmt(txn, s.selectSendToDeviceMessagesStmt).QueryContext(ctx, userID, deviceID, to) if err != nil { return } defer internal.CloseAndLogIfError(ctx, rows, "SelectSendToDeviceMessages: rows.close() failed") for rows.Next() { - var id types.SendToDeviceNID + var id types.StreamPosition var userID, deviceID, content string if err = rows.Scan(&id, &userID, &deviceID, &content); err != nil { return @@ -118,11 +118,11 @@ func (s *sendToDeviceStatements) SelectSendToDeviceMessages( DeviceID: deviceID, } if err = json.Unmarshal([]byte(content), &event.SendToDeviceEvent); err != nil { - return + continue } events = append(events, event) - if types.StreamPosition(id) > lastPos { - lastPos = types.StreamPosition(id) + if id > lastPos { + lastPos = id } } diff --git a/syncapi/storage/shared/syncserver.go b/syncapi/storage/shared/syncserver.go index b2cd24fa1..2771405c1 100644 --- a/syncapi/storage/shared/syncserver.go +++ b/syncapi/storage/shared/syncserver.go @@ -29,6 +29,7 @@ import ( "github.com/matrix-org/dendrite/syncapi/storage/tables" "github.com/matrix-org/dendrite/syncapi/types" "github.com/matrix-org/gomatrixserverlib" + "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus" ) @@ -893,7 +894,7 @@ func (d *Database) StoreNewSendForDeviceMessage( if err != nil { return 0, err } - return + return newPos, nil } func (d *Database) SendToDeviceUpdatesForSync( @@ -902,14 +903,14 @@ func (d *Database) SendToDeviceUpdatesForSync( from, to types.StreamPosition, ) (types.StreamPosition, []types.SendToDeviceEvent, error) { // First of all, get our send-to-device updates for this user. - lastPos, events, err := d.SendToDevice.SelectSendToDeviceMessages(ctx, nil, userID, deviceID, from, to) + lastPos, events, err := d.SendToDevice.SelectSendToDeviceMessages(ctx, nil, userID, deviceID, to) if err != nil { - return 0, nil, fmt.Errorf("d.SendToDevice.SelectSendToDeviceMessages: %w", err) + return from, nil, fmt.Errorf("d.SendToDevice.SelectSendToDeviceMessages: %w", err) } // If there's nothing to do then stop here. if len(events) == 0 { - return 0, nil, nil + return to, nil, nil } // If we've advanced past this stream position for this @@ -917,7 +918,7 @@ func (d *Database) SendToDeviceUpdatesForSync( if err = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error { return d.SendToDevice.DeleteSendToDeviceMessages(ctx, txn, userID, deviceID, from) }); err != nil { - return 0, nil, fmt.Errorf("d.Writer.Do: %w", err) + logrus.WithError(err).Errorf("Failed to clean up old send-to-device messages for user %q device %q", userID, deviceID) } return lastPos, events, nil diff --git a/syncapi/storage/sqlite3/send_to_device_table.go b/syncapi/storage/sqlite3/send_to_device_table.go index 24838e70b..45ec026bd 100644 --- a/syncapi/storage/sqlite3/send_to_device_table.go +++ b/syncapi/storage/sqlite3/send_to_device_table.go @@ -47,7 +47,7 @@ const insertSendToDeviceMessageSQL = ` const selectSendToDeviceMessagesSQL = ` SELECT id, user_id, device_id, content FROM syncapi_send_to_device - WHERE user_id = $1 AND device_id = $2 AND id > $3 AND id <= $4 + WHERE user_id = $1 AND device_id = $2 AND id <= $3 ORDER BY id DESC ` @@ -104,16 +104,16 @@ func (s *sendToDeviceStatements) InsertSendToDeviceMessage( } func (s *sendToDeviceStatements) SelectSendToDeviceMessages( - ctx context.Context, txn *sql.Tx, userID, deviceID string, from, to types.StreamPosition, + ctx context.Context, txn *sql.Tx, userID, deviceID string, to types.StreamPosition, ) (lastPos types.StreamPosition, events []types.SendToDeviceEvent, err error) { - rows, err := sqlutil.TxStmt(txn, s.selectSendToDeviceMessagesStmt).QueryContext(ctx, userID, deviceID, from, to) + rows, err := sqlutil.TxStmt(txn, s.selectSendToDeviceMessagesStmt).QueryContext(ctx, userID, deviceID, to) if err != nil { return } defer internal.CloseAndLogIfError(ctx, rows, "SelectSendToDeviceMessages: rows.close() failed") for rows.Next() { - var id types.SendToDeviceNID + var id types.StreamPosition var userID, deviceID, content string if err = rows.Scan(&id, &userID, &deviceID, &content); err != nil { return @@ -124,11 +124,11 @@ func (s *sendToDeviceStatements) SelectSendToDeviceMessages( DeviceID: deviceID, } if err = json.Unmarshal([]byte(content), &event.SendToDeviceEvent); err != nil { - return + continue } events = append(events, event) - if types.StreamPosition(id) > lastPos { - lastPos = types.StreamPosition(id) + if id > lastPos { + lastPos = id } } diff --git a/syncapi/storage/tables/interface.go b/syncapi/storage/tables/interface.go index e9b12d35e..94070655c 100644 --- a/syncapi/storage/tables/interface.go +++ b/syncapi/storage/tables/interface.go @@ -147,8 +147,8 @@ type BackwardsExtremities interface { // sync response, as the client is seemingly trying to repeat the same /sync. type SendToDevice interface { InsertSendToDeviceMessage(ctx context.Context, txn *sql.Tx, userID, deviceID, content string) (pos types.StreamPosition, err error) - SelectSendToDeviceMessages(ctx context.Context, txn *sql.Tx, userID, deviceID string, from, to types.StreamPosition) (lastPos types.StreamPosition, events []types.SendToDeviceEvent, err error) - DeleteSendToDeviceMessages(ctx context.Context, txn *sql.Tx, userID, deviceID string, pos types.StreamPosition) (err error) + SelectSendToDeviceMessages(ctx context.Context, txn *sql.Tx, userID, deviceID string, to types.StreamPosition) (lastPos types.StreamPosition, events []types.SendToDeviceEvent, err error) + DeleteSendToDeviceMessages(ctx context.Context, txn *sql.Tx, userID, deviceID string, from types.StreamPosition) (err error) SelectMaxSendToDeviceMessageID(ctx context.Context, txn *sql.Tx) (id int64, err error) } diff --git a/syncapi/types/types.go b/syncapi/types/types.go index 8911749fb..35c1fe6ec 100644 --- a/syncapi/types/types.go +++ b/syncapi/types/types.go @@ -492,11 +492,9 @@ func NewLeaveResponse() *LeaveResponse { return &res } -type SendToDeviceNID int - type SendToDeviceEvent struct { gomatrixserverlib.SendToDeviceEvent - ID SendToDeviceNID + ID StreamPosition UserID string DeviceID string }