hoernschen/dendrite
1
0
Fork 0
mirror of https://github.com/hoernschen/dendrite.git synced 2025-04-19 18:23:40 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

(broken) rewrite to use SelectPeeksInRange rather than MarkPeeksAsOld for idempotency

Browse source
This commit is contained in:
Matthew Hodgson 2020-09-08 11:05:54 +01:00
parent 7f41f39684
commit 9f2bc6231f
5 changed files with 54 additions and 106 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

36
syncapi/storage/postgres/peeks_table.go
View file

@ -31,7 +31,6 @@ CREATE TABLE IF NOT EXISTS syncapi_peeks (
room_id TEXT NOT NULL, room_id TEXT NOT NULL,
user_id TEXT NOT NULL, user_id TEXT NOT NULL,
device_id TEXT NOT NULL, device_id TEXT NOT NULL,
new BOOL NOT NULL DEFAULT true,
deleted BOOL NOT NULL DEFAULT false, deleted BOOL NOT NULL DEFAULT false,
-- When the peek was created in UNIX epoch ms. -- When the peek was created in UNIX epoch ms.
creation_ts BIGINT NOT NULL creation_ts BIGINT NOT NULL
@ -52,15 +51,14 @@ const deletePeekSQL = "" +
const deletePeeksSQL = "" + const deletePeeksSQL = "" +
"UPDATE syncapi_peeks SET deleted=true, id=nextval('syncapi_stream_id') WHERE room_id = $1 AND user_id = $2 RETURNING id" "UPDATE syncapi_peeks SET deleted=true, id=nextval('syncapi_stream_id') WHERE room_id = $1 AND user_id = $2 RETURNING id"
const selectPeeksSQL = "" + // we care about all the peeks which were created in this range, deleted in this range,
"SELECT room_id, new FROM syncapi_peeks WHERE user_id = $1 AND device_id = $2 AND deleted=false" // or were created before this range but haven't been deleted yet.
const selectPeeksInRangeSQL = "" +
"SELECT room_id, deleted, (id > $3 AND id <= $4) AS new FROM syncapi_peeks WHERE user_id = $1 AND device_id = $2 AND ((id <= $3 AND NOT deleted) OR new)"
const selectPeekingDevicesSQL = "" + const selectPeekingDevicesSQL = "" +
"SELECT room_id, user_id, device_id FROM syncapi_peeks WHERE deleted=false" "SELECT room_id, user_id, device_id FROM syncapi_peeks WHERE deleted=false"
const markPeeksAsOldSQL = "" +
"UPDATE syncapi_peeks SET id=nextval('syncapi_stream_id'), new=false WHERE user_id = $1 AND device_id = $2 AND deleted=false RETURNING id"
const selectMaxPeekIDSQL = "" + const selectMaxPeekIDSQL = "" +
"SELECT MAX(id) FROM syncapi_peeks" "SELECT MAX(id) FROM syncapi_peeks"
@ -69,9 +67,8 @@ type peekStatements struct {
insertPeekStmt *sql.Stmt insertPeekStmt *sql.Stmt
deletePeekStmt *sql.Stmt deletePeekStmt *sql.Stmt
deletePeeksStmt *sql.Stmt deletePeeksStmt *sql.Stmt
selectPeeksStmt *sql.Stmt selectPeeksInRangeStmt *sql.Stmt
selectPeekingDevicesStmt *sql.Stmt selectPeekingDevicesStmt *sql.Stmt
markPeeksAsOldStmt *sql.Stmt
selectMaxPeekIDStmt *sql.Stmt selectMaxPeekIDStmt *sql.Stmt
} }
@ -92,15 +89,12 @@ func NewPostgresPeeksTable(db *sql.DB) (tables.Peeks, error) {
if s.deletePeeksStmt, err = db.Prepare(deletePeeksSQL); err != nil { if s.deletePeeksStmt, err = db.Prepare(deletePeeksSQL); err != nil {
return nil, err return nil, err
} }
if s.selectPeeksStmt, err = db.Prepare(selectPeeksSQL); err != nil { if s.selectPeeksInRangeStmt, err = db.Prepare(selectPeeksInRangeSQL); err != nil {
return nil, err return nil, err
} }
if s.selectPeekingDevicesStmt, err = db.Prepare(selectPeekingDevicesSQL); err != nil { if s.selectPeekingDevicesStmt, err = db.Prepare(selectPeekingDevicesSQL); err != nil {
return nil, err return nil, err
} }
if s.markPeeksAsOldStmt, err = db.Prepare(markPeeksAsOldSQL); err != nil {
return nil, err
}
if s.selectMaxPeekIDStmt, err = db.Prepare(selectMaxPeekIDSQL); err != nil { if s.selectMaxPeekIDStmt, err = db.Prepare(selectMaxPeekIDSQL); err != nil {
return nil, err return nil, err
} }
@ -132,18 +126,18 @@ func (s *peekStatements) DeletePeeks(
return return
} }
func (s *peekStatements) SelectPeeks( func (s *peekStatements) SelectPeeksInRange(
ctx context.Context, txn *sql.Tx, userID, deviceID string, ctx context.Context, txn *sql.Tx, userID, deviceID string, r types.Range,
) (peeks []types.Peek, err error) { ) (peeks []types.Peek, err error) {
rows, err := sqlutil.TxStmt(txn, s.selectPeeksStmt).QueryContext(ctx, userID, deviceID) rows, err := sqlutil.TxStmt(txn, s.selectPeeksInRangeStmt).QueryContext(ctx, userID, deviceID, r.Low(), r.High())
if err != nil { if err != nil {
return return
} }
defer internal.CloseAndLogIfError(ctx, rows, "SelectPeeks: rows.close() failed") defer internal.CloseAndLogIfError(ctx, rows, "SelectPeeksInRange: rows.close() failed")
for rows.Next() { for rows.Next() {
peek := types.Peek{} peek := types.Peek{}
if err = rows.Scan(&peek.RoomID, &peek.New); err != nil { if err = rows.Scan(&peek.RoomID, &peek.Deleted, &peek.New); err != nil {
return return
} }
peeks = append(peeks, peek) peeks = append(peeks, peek)
@ -152,14 +146,6 @@ func (s *peekStatements) SelectPeeks(
return peeks, rows.Err() return peeks, rows.Err()
} }
func (s *peekStatements) MarkPeeksAsOld(
ctx context.Context, txn *sql.Tx, userID, deviceID string,
) (streamPos types.StreamPosition, err error) {
stmt := sqlutil.TxStmt(txn, s.markPeeksAsOldStmt)
err = stmt.QueryRowContext(ctx, userID, deviceID).Scan(&streamPos)
return
}
func (s *peekStatements) SelectPeekingDevices( func (s *peekStatements) SelectPeekingDevices(
ctx context.Context, ctx context.Context,
) (peekingDevices map[string][]types.PeekingDevice, err error) { ) (peekingDevices map[string][]types.PeekingDevice, err error) {

35
syncapi/storage/shared/syncserver.go
View file

@ -694,11 +694,12 @@ func (d *Database) getResponseWithPDUsForCompleteSync(
} }
// Add peeked rooms. // Add peeked rooms.
peeks, err := d.Peeks.SelectPeeks(ctx, txn, userID, deviceID) peeks, err := d.Peeks.SelectPeeksInRange(ctx, txn, userID, deviceID, r)
if err != nil { if err != nil {
return return
} }
for _, peek := range peeks { for _, peek := range peeks {
if !peek.Deleted {
var jr *types.JoinResponse var jr *types.JoinResponse
jr, err = d.getJoinResponseForCompleteSync( jr, err = d.getJoinResponseForCompleteSync(
ctx, txn, peek.RoomID, r, &stateFilter, numRecentEventsPerRoom, ctx, txn, peek.RoomID, r, &stateFilter, numRecentEventsPerRoom,
@ -708,6 +709,7 @@ func (d *Database) getResponseWithPDUsForCompleteSync(
} }
res.Rooms.Peek[peek.RoomID] = *jr res.Rooms.Peek[peek.RoomID] = *jr
} }
}
if err = d.addInvitesToResponse(ctx, txn, userID, r, res); err != nil { if err = d.addInvitesToResponse(ctx, txn, userID, r, res); err != nil {
return return
@ -1034,13 +1036,12 @@ func (d *Database) getStateDeltas(
// find out which rooms this user is peeking, if any. // find out which rooms this user is peeking, if any.
// We do this before joins so any peeks get overwritten // We do this before joins so any peeks get overwritten
peeks, err := d.Peeks.SelectPeeks(ctx, txn, userID, device.ID) peeks, err := d.Peeks.SelectPeeksInRange(ctx, txn, userID, device.ID, r)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }
// add peek blocks // add peek blocks
newPeeks := false
for _, peek := range peeks { for _, peek := range peeks {
if peek.New { if peek.New {
// send full room state down instead of a delta // send full room state down instead of a delta
@ -1050,24 +1051,14 @@ func (d *Database) getStateDeltas(
return nil, nil, err return nil, nil, err
} }
state[peek.RoomID] = s state[peek.RoomID] = s
newPeeks = true
} }
if !peek.Deleted {
deltas = append(deltas, stateDelta{ deltas = append(deltas, stateDelta{
membership: gomatrixserverlib.Peek, membership: gomatrixserverlib.Peek,
stateEvents: d.StreamEventsToEvents(device, state[peek.RoomID]), stateEvents: d.StreamEventsToEvents(device, state[peek.RoomID]),
roomID: peek.RoomID, roomID: peek.RoomID,
}) })
} }
if newPeeks {
err = d.Writer.Do(d.DB, txn, func(txn *sql.Tx) error {
_, err := d.Peeks.MarkPeeksAsOld(ctx, txn, userID, device.ID)
return err
})
if err != nil {
return nil, nil, err
}
} }
// handle newly joined rooms and non-joined rooms // handle newly joined rooms and non-joined rooms
@ -1130,17 +1121,14 @@ func (d *Database) getStateDeltasForFullStateSync(
// Use a reasonable initial capacity // Use a reasonable initial capacity
deltas := make(map[string]stateDelta) deltas := make(map[string]stateDelta)
peeks, err := d.Peeks.SelectPeeks(ctx, txn, userID, device.ID) peeks, err := d.Peeks.SelectPeeksInRange(ctx, txn, userID, device.ID, r)
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }
// Add full states for all peeking rooms // Add full states for all peeking rooms
newPeeks := false
for _, peek := range peeks { for _, peek := range peeks {
if peek.New { if !peek.Deleted {
newPeeks = true
}
s, stateErr := d.currentStateStreamEventsForRoom(ctx, txn, peek.RoomID, stateFilter) s, stateErr := d.currentStateStreamEventsForRoom(ctx, txn, peek.RoomID, stateFilter)
if stateErr != nil { if stateErr != nil {
return nil, nil, stateErr return nil, nil, stateErr
@ -1151,15 +1139,6 @@ func (d *Database) getStateDeltasForFullStateSync(
roomID: peek.RoomID, roomID: peek.RoomID,
} }
} }
if newPeeks {
err = d.Writer.Do(d.DB, txn, func(txn *sql.Tx) error {
_, err := d.Peeks.MarkPeeksAsOld(ctx, txn, userID, device.ID)
return err
})
if err != nil {
return nil, nil, err
}
} }
// Get all the state events ever between these two positions // Get all the state events ever between these two positions

39
syncapi/storage/sqlite3/peeks_table.go
View file

@ -31,7 +31,6 @@ CREATE TABLE IF NOT EXISTS syncapi_peeks (
room_id TEXT NOT NULL, room_id TEXT NOT NULL,
user_id TEXT NOT NULL, user_id TEXT NOT NULL,
device_id TEXT NOT NULL, device_id TEXT NOT NULL,
new BOOL NOT NULL DEFAULT true,
deleted BOOL NOT NULL DEFAULT false, deleted BOOL NOT NULL DEFAULT false,
-- When the peek was created in UNIX epoch ms. -- When the peek was created in UNIX epoch ms.
creation_ts INTEGER NOT NULL creation_ts INTEGER NOT NULL
@ -52,15 +51,14 @@ const deletePeekSQL = "" +
const deletePeeksSQL = "" + const deletePeeksSQL = "" +
"UPDATE syncapi_peeks SET deleted=true, id=$1 WHERE room_id = $2 AND user_id = $3" "UPDATE syncapi_peeks SET deleted=true, id=$1 WHERE room_id = $2 AND user_id = $3"
const selectPeeksSQL = "" + // we care about all the peeks which were created in this range, deleted in this range,
"SELECT room_id, new FROM syncapi_peeks WHERE user_id = $1 AND device_id = $2 AND deleted=false" // or were created before this range but haven't been deleted yet.
const selectPeeksInRangeSQL = "" +
"SELECT room_id, deleted, (id > $3 AND id <= $4) AS new FROM syncapi_peeks WHERE user_id = $1 AND device_id = $2 AND ((id <= $3 AND NOT deleted) OR new)"
const selectPeekingDevicesSQL = "" + const selectPeekingDevicesSQL = "" +
"SELECT room_id, user_id, device_id FROM syncapi_peeks WHERE deleted=false" "SELECT room_id, user_id, device_id FROM syncapi_peeks WHERE deleted=false"
const markPeeksAsOldSQL = "" +
"UPDATE syncapi_peeks SET new=false, id=$1 WHERE user_id = $2 AND device_id = $3 AND deleted=false"
const selectMaxPeekIDSQL = "" + const selectMaxPeekIDSQL = "" +
"SELECT MAX(id) FROM syncapi_peeks" "SELECT MAX(id) FROM syncapi_peeks"
@ -70,9 +68,8 @@ type peekStatements struct {
insertPeekStmt *sql.Stmt insertPeekStmt *sql.Stmt
deletePeekStmt *sql.Stmt deletePeekStmt *sql.Stmt
deletePeeksStmt *sql.Stmt deletePeeksStmt *sql.Stmt
selectPeeksStmt *sql.Stmt selectPeeksInRangeStmt *sql.Stmt
selectPeekingDevicesStmt *sql.Stmt selectPeekingDevicesStmt *sql.Stmt
markPeeksAsOldStmt *sql.Stmt
selectMaxPeekIDStmt *sql.Stmt selectMaxPeekIDStmt *sql.Stmt
} }
@ -94,15 +91,12 @@ func NewSqlitePeeksTable(db *sql.DB, streamID *streamIDStatements) (tables.Peeks
if s.deletePeeksStmt, err = db.Prepare(deletePeeksSQL); err != nil { if s.deletePeeksStmt, err = db.Prepare(deletePeeksSQL); err != nil {
return nil, err return nil, err
} }
if s.selectPeeksStmt, err = db.Prepare(selectPeeksSQL); err != nil { if s.selectPeeksInRangeStmt, err = db.Prepare(selectPeeksInRangeSQL); err != nil {
return nil, err return nil, err
} }
if s.selectPeekingDevicesStmt, err = db.Prepare(selectPeekingDevicesSQL); err != nil { if s.selectPeekingDevicesStmt, err = db.Prepare(selectPeekingDevicesSQL); err != nil {
return nil, err return nil, err
} }
if s.markPeeksAsOldStmt, err = db.Prepare(markPeeksAsOldSQL); err != nil {
return nil, err
}
if s.selectMaxPeekIDStmt, err = db.Prepare(selectMaxPeekIDSQL); err != nil { if s.selectMaxPeekIDStmt, err = db.Prepare(selectMaxPeekIDSQL); err != nil {
return nil, err return nil, err
} }
@ -143,18 +137,18 @@ func (s *peekStatements) DeletePeeks(
return return
} }
func (s *peekStatements) SelectPeeks( func (s *peekStatements) SelectPeeksInRange(
ctx context.Context, txn *sql.Tx, userID, deviceID string, ctx context.Context, txn *sql.Tx, userID, deviceID string, r types.Range,
) (peeks []types.Peek, err error) { ) (peeks []types.Peek, err error) {
rows, err := sqlutil.TxStmt(txn, s.selectPeeksStmt).QueryContext(ctx, userID, deviceID) rows, err := sqlutil.TxStmt(txn, s.selectPeeksInRangeStmt).QueryContext(ctx, userID, deviceID, r.Low(), r.High())
if err != nil { if err != nil {
return return
} }
defer internal.CloseAndLogIfError(ctx, rows, "SelectPeeks: rows.close() failed") defer internal.CloseAndLogIfError(ctx, rows, "SelectPeeksInRange: rows.close() failed")
for rows.Next() { for rows.Next() {
peek := types.Peek{} peek := types.Peek{}
if err = rows.Scan(&peek.RoomID, &peek.New); err != nil { if err = rows.Scan(&peek.RoomID, &peek.Deleted, &peek.New); err != nil {
return return
} }
peeks = append(peeks, peek) peeks = append(peeks, peek)
@ -163,17 +157,6 @@ func (s *peekStatements) SelectPeeks(
return peeks, rows.Err() return peeks, rows.Err()
} }
func (s *peekStatements) MarkPeeksAsOld(
ctx context.Context, txn *sql.Tx, userID, deviceID string,
) (streamPos types.StreamPosition, err error) {
streamPos, err = s.streamIDStatements.nextStreamID(ctx, txn)
if err != nil {
return
}
_, err = sqlutil.TxStmt(txn, s.markPeeksAsOldStmt).ExecContext(ctx, streamPos, userID, deviceID)
return
}
func (s *peekStatements) SelectPeekingDevices( func (s *peekStatements) SelectPeekingDevices(
ctx context.Context, ctx context.Context,
) (peekingDevices map[string][]types.PeekingDevice, err error) { ) (peekingDevices map[string][]types.PeekingDevice, err error) {

3
syncapi/storage/tables/interface.go
View file

@ -43,9 +43,8 @@ type Peeks interface {
InsertPeek(ctx context.Context, txn *sql.Tx, roomID, userID, deviceID string) (streamPos types.StreamPosition, err error) InsertPeek(ctx context.Context, txn *sql.Tx, roomID, userID, deviceID string) (streamPos types.StreamPosition, err error)
DeletePeek(ctx context.Context, txn *sql.Tx, roomID, userID, deviceID string) (streamPos types.StreamPosition, err error) DeletePeek(ctx context.Context, txn *sql.Tx, roomID, userID, deviceID string) (streamPos types.StreamPosition, err error)
DeletePeeks(ctx context.Context, txn *sql.Tx, roomID, userID string) (streamPos types.StreamPosition, err error) DeletePeeks(ctx context.Context, txn *sql.Tx, roomID, userID string) (streamPos types.StreamPosition, err error)
SelectPeeks(ctxt context.Context, txn *sql.Tx, userID, deviceID string) (peeks []types.Peek, err error) SelectPeeksInRange(ctxt context.Context, txn *sql.Tx, userID, deviceID string, r types.Range) (peeks []types.Peek, err error)
SelectPeekingDevices(ctxt context.Context) (peekingDevices map[string][]types.PeekingDevice, err error) SelectPeekingDevices(ctxt context.Context) (peekingDevices map[string][]types.PeekingDevice, err error)
MarkPeeksAsOld(ctxt context.Context, txn *sql.Tx, userID, deviceID string) (streamPos types.StreamPosition, err error)
SelectMaxPeekID(ctx context.Context, txn *sql.Tx) (id int64, err error) SelectMaxPeekID(ctx context.Context, txn *sql.Tx) (id int64, err error)
} }

1
syncapi/types/types.go
View file

@ -518,4 +518,5 @@ type PeekingDevice struct {
type Peek struct { type Peek struct {
RoomID string RoomID string
New bool New bool
Deleted bool
} }