Mirrors/dendrite
2
0
Fork 0
mirror of https://github.com/matrix-org/dendrite synced 2024-09-21 07:02:01 +00:00
Code Issues Projects Releases Packages Wiki Activity

wip sqlite merges; database is locked errors to investigate and failing tests

Browse source
This commit is contained in:
Kegan Dougal 2020-05-14 13:09:27 +01:00
parent 08d0eda31c
commit 26cbfc5b75
10 changed files with 98 additions and 1058 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
syncapi/storage/postgres/account_data_table.go
View file

@ -143,8 +143,7 @@ func (s *accountDataStatements) SelectMaxAccountDataID(
ctx context.Context, txn *sql.Tx, ctx context.Context, txn *sql.Tx,
) (id int64, err error) { ) (id int64, err error) {
var nullableID sql.NullInt64 var nullableID sql.NullInt64
stmt := common.TxStmt(txn, s.selectMaxAccountDataIDStmt) err = txn.Stmt(s.selectMaxAccountDataIDStmt).QueryRowContext(ctx).Scan(&nullableID)
err = stmt.QueryRowContext(ctx).Scan(&nullableID)
if nullableID.Valid { if nullableID.Valid {
id = nullableID.Int64 id = nullableID.Int64
} }

18
syncapi/storage/postgres/output_room_events_topology_table.go
View file

@ -115,9 +115,9 @@ func NewPostgresTopologyTable(db *sql.DB) (tables.Topology, error) {
// InsertEventInTopology inserts the given event in the room's topology, based // InsertEventInTopology inserts the given event in the room's topology, based
// on the event's depth. // on the event's depth.
func (s *outputRoomEventsTopologyStatements) InsertEventInTopology( func (s *outputRoomEventsTopologyStatements) InsertEventInTopology(
ctx context.Context, event *gomatrixserverlib.HeaderedEvent, pos types.StreamPosition, ctx context.Context, txn *sql.Tx, event *gomatrixserverlib.HeaderedEvent, pos types.StreamPosition,
) (err error) { ) (err error) {
_, err = s.insertEventInTopologyStmt.ExecContext( _, err = common.TxStmt(txn, s.insertEventInTopologyStmt).ExecContext(
ctx, event.EventID(), event.Depth(), event.RoomID(), pos, ctx, event.EventID(), event.Depth(), event.RoomID(), pos,
) )
return return
@ -127,7 +127,7 @@ func (s *outputRoomEventsTopologyStatements) InsertEventInTopology(
// given range in a given room's topological order. // given range in a given room's topological order.
// Returns an empty slice if no events match the given range. // Returns an empty slice if no events match the given range.
func (s *outputRoomEventsTopologyStatements) SelectEventIDsInRange( func (s *outputRoomEventsTopologyStatements) SelectEventIDsInRange(
ctx context.Context, roomID string, fromPos, toPos, toMicroPos types.StreamPosition, ctx context.Context, txn *sql.Tx, roomID string, fromPos, toPos, toMicroPos types.StreamPosition,
limit int, chronologicalOrder bool, limit int, chronologicalOrder bool,
) (eventIDs []string, err error) { ) (eventIDs []string, err error) {
// Decide on the selection's order according to whether chronological order // Decide on the selection's order according to whether chronological order
@ -164,26 +164,26 @@ func (s *outputRoomEventsTopologyStatements) SelectEventIDsInRange(
// SelectPositionInTopology returns the position of a given event in the // SelectPositionInTopology returns the position of a given event in the
// topology of the room it belongs to. // topology of the room it belongs to.
func (s *outputRoomEventsTopologyStatements) SelectPositionInTopology( func (s *outputRoomEventsTopologyStatements) SelectPositionInTopology(
ctx context.Context, eventID string, ctx context.Context, txn *sql.Tx, eventID string,
) (pos, spos types.StreamPosition, err error) { ) (pos, spos types.StreamPosition, err error) {
err = s.selectPositionInTopologyStmt.QueryRowContext(ctx, eventID).Scan(&pos, &spos) err = common.TxStmt(txn, s.selectPositionInTopologyStmt).QueryRowContext(ctx, eventID).Scan(&pos, &spos)
return return
} }
func (s *outputRoomEventsTopologyStatements) SelectMaxPositionInTopology( func (s *outputRoomEventsTopologyStatements) SelectMaxPositionInTopology(
ctx context.Context, roomID string, ctx context.Context, txn *sql.Tx, roomID string,
) (pos types.StreamPosition, spos types.StreamPosition, err error) { ) (pos types.StreamPosition, spos types.StreamPosition, err error) {
err = s.selectMaxPositionInTopologyStmt.QueryRowContext(ctx, roomID).Scan(&pos, &spos) err = common.TxStmt(txn, s.selectMaxPositionInTopologyStmt).QueryRowContext(ctx, roomID).Scan(&pos, &spos)
return return
} }
// SelectEventIDsFromPosition returns the IDs of all events that have a given // SelectEventIDsFromPosition returns the IDs of all events that have a given
// position in the topology of a given room. // position in the topology of a given room.
func (s *outputRoomEventsTopologyStatements) SelectEventIDsFromPosition( func (s *outputRoomEventsTopologyStatements) SelectEventIDsFromPosition(
ctx context.Context, roomID string, pos types.StreamPosition, ctx context.Context, txn *sql.Tx, roomID string, pos types.StreamPosition,
) (eventIDs []string, err error) { ) (eventIDs []string, err error) {
// Query the event IDs. // Query the event IDs.
rows, err := s.selectEventIDsFromPositionStmt.QueryContext(ctx, roomID, pos) rows, err := common.TxStmt(txn, s.selectEventIDsFromPositionStmt).QueryContext(ctx, roomID, pos)
if err == sql.ErrNoRows { if err == sql.ErrNoRows {
// If no event matched the request, return an empty slice. // If no event matched the request, return an empty slice.
return []string{}, nil return []string{}, nil

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

@ -260,7 +260,7 @@ func (d *Database) WriteEvent(
} }
pduPosition = pos pduPosition = pos
if err = d.Topology.InsertEventInTopology(ctx, ev, pos); err != nil { if err = d.Topology.InsertEventInTopology(ctx, txn, ev, pos); err != nil {
return err return err
} }
@ -337,7 +337,7 @@ func (d *Database) GetEventsInTopologicalRange(
// Select the event IDs from the defined range. // Select the event IDs from the defined range.
var eIDs []string var eIDs []string
eIDs, err = d.Topology.SelectEventIDsInRange( eIDs, err = d.Topology.SelectEventIDsInRange(
ctx, roomID, backwardLimit, forwardLimit, forwardMicroLimit, limit, !backwardOrdering, ctx, nil, roomID, backwardLimit, forwardLimit, forwardMicroLimit, limit, !backwardOrdering,
) )
if err != nil { if err != nil {
return return
@ -348,8 +348,12 @@ func (d *Database) GetEventsInTopologicalRange(
return return
} }
func (d *Database) SyncPosition(ctx context.Context) (types.StreamingToken, error) { func (d *Database) SyncPosition(ctx context.Context) (st types.StreamingToken, err error) {
return d.syncPositionTx(ctx, nil) err = common.WithTransaction(d.DB, func(txn *sql.Tx) error {
st, err = d.syncPositionTx(ctx, txn)
return err
})
return
} }
func (d *Database) BackwardExtremitiesForRoom( func (d *Database) BackwardExtremitiesForRoom(
@ -361,13 +365,13 @@ func (d *Database) BackwardExtremitiesForRoom(
func (d *Database) MaxTopologicalPosition( func (d *Database) MaxTopologicalPosition(
ctx context.Context, roomID string, ctx context.Context, roomID string,
) (depth types.StreamPosition, stream types.StreamPosition, err error) { ) (depth types.StreamPosition, stream types.StreamPosition, err error) {
return d.Topology.SelectMaxPositionInTopology(ctx, roomID) return d.Topology.SelectMaxPositionInTopology(ctx, nil, roomID)
} }
func (d *Database) EventsAtTopologicalPosition( func (d *Database) EventsAtTopologicalPosition(
ctx context.Context, roomID string, pos types.StreamPosition, ctx context.Context, roomID string, pos types.StreamPosition,
) ([]types.StreamEvent, error) { ) ([]types.StreamEvent, error) {
eIDs, err := d.Topology.SelectEventIDsFromPosition(ctx, roomID, pos) eIDs, err := d.Topology.SelectEventIDsFromPosition(ctx, nil, roomID, pos)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -378,7 +382,7 @@ func (d *Database) EventsAtTopologicalPosition(
func (d *Database) EventPositionInTopology( func (d *Database) EventPositionInTopology(
ctx context.Context, eventID string, ctx context.Context, eventID string,
) (depth types.StreamPosition, stream types.StreamPosition, err error) { ) (depth types.StreamPosition, stream types.StreamPosition, err error) {
return d.Topology.SelectPositionInTopology(ctx, eventID) return d.Topology.SelectPositionInTopology(ctx, nil, eventID)
} }
func (d *Database) syncPositionTx( func (d *Database) syncPositionTx(
@ -618,7 +622,7 @@ func (d *Database) getResponseWithPDUsForCompleteSync(
var prevBatchStr string var prevBatchStr string
if len(recentStreamEvents) > 0 { if len(recentStreamEvents) > 0 {
var backwardTopologyPos, backwardStreamPos types.StreamPosition var backwardTopologyPos, backwardStreamPos types.StreamPosition
backwardTopologyPos, backwardStreamPos, err = d.Topology.SelectPositionInTopology(ctx, recentStreamEvents[0].EventID()) backwardTopologyPos, backwardStreamPos, err = d.Topology.SelectPositionInTopology(ctx, nil, recentStreamEvents[0].EventID())
if err != nil { if err != nil {
return return
} }
@ -702,9 +706,9 @@ func (d *Database) addInvitesToResponse(
func (d *Database) getBackwardTopologyPos( func (d *Database) getBackwardTopologyPos(
ctx context.Context, ctx context.Context,
events []types.StreamEvent, events []types.StreamEvent,
) (pos, spos types.StreamPosition) { ) (pos, spos types.StreamPosition, err error) {
if len(events) > 0 { if len(events) > 0 {
pos, spos, _ = d.Topology.SelectPositionInTopology(ctx, events[0].EventID()) pos, spos, err = d.Topology.SelectPositionInTopology(ctx, nil, events[0].EventID())
} }
if pos-1 <= 0 { if pos-1 <= 0 {
pos = types.StreamPosition(1) pos = types.StreamPosition(1)
@ -743,7 +747,10 @@ func (d *Database) addRoomDeltaToResponse(
} }
recentEvents := d.StreamEventsToEvents(device, recentStreamEvents) recentEvents := d.StreamEventsToEvents(device, recentStreamEvents)
delta.stateEvents = removeDuplicates(delta.stateEvents, recentEvents) // roll back delta.stateEvents = removeDuplicates(delta.stateEvents, recentEvents) // roll back
backwardTopologyPos, backwardStreamPos := d.getBackwardTopologyPos(ctx, recentStreamEvents) backwardTopologyPos, backwardStreamPos, err := d.getBackwardTopologyPos(ctx, recentStreamEvents)
if err != nil {
return err
}
prevBatch := types.NewTopologyToken( prevBatch := types.NewTopologyToken(
backwardTopologyPos, backwardStreamPos, backwardTopologyPos, backwardStreamPos,
) )

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

@ -22,6 +22,7 @@ import (
"strings" "strings"
"github.com/matrix-org/dendrite/common" "github.com/matrix-org/dendrite/common"
"github.com/matrix-org/dendrite/syncapi/storage/tables"
"github.com/matrix-org/dendrite/syncapi/types" "github.com/matrix-org/dendrite/syncapi/types"
"github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/gomatrixserverlib"
) )
@ -91,35 +92,37 @@ type currentRoomStateStatements struct {
selectStateEventStmt *sql.Stmt selectStateEventStmt *sql.Stmt
} }
func (s *currentRoomStateStatements) prepare(db *sql.DB, streamID *streamIDStatements) (err error) { func NewSqliteCurrentRoomStateTable(db *sql.DB, streamID *streamIDStatements) (tables.CurrentRoomState, error) {
s.streamIDStatements = streamID s := &currentRoomStateStatements{
_, err = db.Exec(currentRoomStateSchema) streamIDStatements: streamID,
}
_, err := db.Exec(currentRoomStateSchema)
if err != nil { if err != nil {
return return nil, err
} }
if s.upsertRoomStateStmt, err = db.Prepare(upsertRoomStateSQL); err != nil { if s.upsertRoomStateStmt, err = db.Prepare(upsertRoomStateSQL); err != nil {
return return nil, err
} }
if s.deleteRoomStateByEventIDStmt, err = db.Prepare(deleteRoomStateByEventIDSQL); err != nil { if s.deleteRoomStateByEventIDStmt, err = db.Prepare(deleteRoomStateByEventIDSQL); err != nil {
return return nil, err
} }
if s.selectRoomIDsWithMembershipStmt, err = db.Prepare(selectRoomIDsWithMembershipSQL); err != nil { if s.selectRoomIDsWithMembershipStmt, err = db.Prepare(selectRoomIDsWithMembershipSQL); err != nil {
return return nil, err
} }
if s.selectCurrentStateStmt, err = db.Prepare(selectCurrentStateSQL); err != nil { if s.selectCurrentStateStmt, err = db.Prepare(selectCurrentStateSQL); err != nil {
return return nil, err
} }
if s.selectJoinedUsersStmt, err = db.Prepare(selectJoinedUsersSQL); err != nil { if s.selectJoinedUsersStmt, err = db.Prepare(selectJoinedUsersSQL); err != nil {
return return nil, err
} }
if s.selectStateEventStmt, err = db.Prepare(selectStateEventSQL); err != nil { if s.selectStateEventStmt, err = db.Prepare(selectStateEventSQL); err != nil {
return return nil, err
} }
return return s, nil
} }
// JoinedMemberLists returns a map of room ID to a list of joined user IDs. // JoinedMemberLists returns a map of room ID to a list of joined user IDs.
func (s *currentRoomStateStatements) selectJoinedUsers( func (s *currentRoomStateStatements) SelectJoinedUsers(
ctx context.Context, ctx context.Context,
) (map[string][]string, error) { ) (map[string][]string, error) {
rows, err := s.selectJoinedUsersStmt.QueryContext(ctx) rows, err := s.selectJoinedUsersStmt.QueryContext(ctx)
@ -143,7 +146,7 @@ func (s *currentRoomStateStatements) selectJoinedUsers(
} }
// SelectRoomIDsWithMembership returns the list of room IDs which have the given user in the given membership state. // SelectRoomIDsWithMembership returns the list of room IDs which have the given user in the given membership state.
func (s *currentRoomStateStatements) selectRoomIDsWithMembership( func (s *currentRoomStateStatements) SelectRoomIDsWithMembership(
ctx context.Context, ctx context.Context,
txn *sql.Tx, txn *sql.Tx,
userID string, userID string,
@ -168,7 +171,7 @@ func (s *currentRoomStateStatements) selectRoomIDsWithMembership(
} }
// CurrentState returns all the current state events for the given room. // CurrentState returns all the current state events for the given room.
func (s *currentRoomStateStatements) selectCurrentState( func (s *currentRoomStateStatements) SelectCurrentState(
ctx context.Context, txn *sql.Tx, roomID string, ctx context.Context, txn *sql.Tx, roomID string,
stateFilterPart *gomatrixserverlib.StateFilter, stateFilterPart *gomatrixserverlib.StateFilter,
) ([]gomatrixserverlib.HeaderedEvent, error) { ) ([]gomatrixserverlib.HeaderedEvent, error) {
@ -189,7 +192,7 @@ func (s *currentRoomStateStatements) selectCurrentState(
return rowsToEvents(rows) return rowsToEvents(rows)
} }
func (s *currentRoomStateStatements) deleteRoomStateByEventID( func (s *currentRoomStateStatements) DeleteRoomStateByEventID(
ctx context.Context, txn *sql.Tx, eventID string, ctx context.Context, txn *sql.Tx, eventID string,
) error { ) error {
stmt := common.TxStmt(txn, s.deleteRoomStateByEventIDStmt) stmt := common.TxStmt(txn, s.deleteRoomStateByEventIDStmt)
@ -197,7 +200,7 @@ func (s *currentRoomStateStatements) deleteRoomStateByEventID(
return err return err
} }
func (s *currentRoomStateStatements) upsertRoomState( func (s *currentRoomStateStatements) UpsertRoomState(
ctx context.Context, txn *sql.Tx, ctx context.Context, txn *sql.Tx,
event gomatrixserverlib.HeaderedEvent, membership *string, addedAt types.StreamPosition, event gomatrixserverlib.HeaderedEvent, membership *string, addedAt types.StreamPosition,
) error { ) error {
@ -231,7 +234,7 @@ func (s *currentRoomStateStatements) upsertRoomState(
return err return err
} }
func (s *currentRoomStateStatements) selectEventsWithEventIDs( func (s *currentRoomStateStatements) SelectEventsWithEventIDs(
ctx context.Context, txn *sql.Tx, eventIDs []string, ctx context.Context, txn *sql.Tx, eventIDs []string,
) ([]types.StreamEvent, error) { ) ([]types.StreamEvent, error) {
iEventIDs := make([]interface{}, len(eventIDs)) iEventIDs := make([]interface{}, len(eventIDs))
@ -264,7 +267,7 @@ func rowsToEvents(rows *sql.Rows) ([]gomatrixserverlib.HeaderedEvent, error) {
return result, nil return result, nil
} }
func (s *currentRoomStateStatements) selectStateEvent( func (s *currentRoomStateStatements) SelectStateEvent(
ctx context.Context, roomID, evType, stateKey string, ctx context.Context, roomID, evType, stateKey string,
) (*gomatrixserverlib.HeaderedEvent, error) { ) (*gomatrixserverlib.HeaderedEvent, error) {
stmt := s.selectStateEventStmt stmt := s.selectStateEventStmt

1
syncapi/storage/sqlite3/output_room_events_table.go
View file

@ -320,7 +320,6 @@ func (s *outputRoomEventsStatements) SelectRecentEvents(
} else { } else {
stmt = common.TxStmt(txn, s.selectRecentEventsStmt) stmt = common.TxStmt(txn, s.selectRecentEventsStmt)
} }
rows, err := stmt.QueryContext(ctx, roomID, fromPos, toPos, limit) rows, err := stmt.QueryContext(ctx, roomID, fromPos, toPos, limit)
if err != nil { if err != nil {
return nil, err return nil, err

32
syncapi/storage/sqlite3/output_room_events_topology_table.go
View file

@ -19,6 +19,7 @@ import (
"database/sql" "database/sql"
"github.com/matrix-org/dendrite/common" "github.com/matrix-org/dendrite/common"
"github.com/matrix-org/dendrite/syncapi/storage/tables"
"github.com/matrix-org/dendrite/syncapi/types" "github.com/matrix-org/dendrite/syncapi/types"
"github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/gomatrixserverlib"
) )
@ -77,35 +78,36 @@ type outputRoomEventsTopologyStatements struct {
selectEventIDsFromPositionStmt *sql.Stmt selectEventIDsFromPositionStmt *sql.Stmt
} }
func (s *outputRoomEventsTopologyStatements) prepare(db *sql.DB) (err error) { func NewSqliteTopologyTable(db *sql.DB) (tables.Topology, error) {
_, err = db.Exec(outputRoomEventsTopologySchema) s := &outputRoomEventsTopologyStatements{}
_, err := db.Exec(outputRoomEventsTopologySchema)
if err != nil { if err != nil {
return return nil, err
} }
if s.insertEventInTopologyStmt, err = db.Prepare(insertEventInTopologySQL); err != nil { if s.insertEventInTopologyStmt, err = db.Prepare(insertEventInTopologySQL); err != nil {
return return nil, err
} }
if s.selectEventIDsInRangeASCStmt, err = db.Prepare(selectEventIDsInRangeASCSQL); err != nil { if s.selectEventIDsInRangeASCStmt, err = db.Prepare(selectEventIDsInRangeASCSQL); err != nil {
return return nil, err
} }
if s.selectEventIDsInRangeDESCStmt, err = db.Prepare(selectEventIDsInRangeDESCSQL); err != nil { if s.selectEventIDsInRangeDESCStmt, err = db.Prepare(selectEventIDsInRangeDESCSQL); err != nil {
return return nil, err
} }
if s.selectPositionInTopologyStmt, err = db.Prepare(selectPositionInTopologySQL); err != nil { if s.selectPositionInTopologyStmt, err = db.Prepare(selectPositionInTopologySQL); err != nil {
return return nil, err
} }
if s.selectMaxPositionInTopologyStmt, err = db.Prepare(selectMaxPositionInTopologySQL); err != nil { if s.selectMaxPositionInTopologyStmt, err = db.Prepare(selectMaxPositionInTopologySQL); err != nil {
return return nil, err
} }
if s.selectEventIDsFromPositionStmt, err = db.Prepare(selectEventIDsFromPositionSQL); err != nil { if s.selectEventIDsFromPositionStmt, err = db.Prepare(selectEventIDsFromPositionSQL); err != nil {
return return nil, err
} }
return return s, nil
} }
// insertEventInTopology inserts the given event in the room's topology, based // insertEventInTopology inserts the given event in the room's topology, based
// on the event's depth. // on the event's depth.
func (s *outputRoomEventsTopologyStatements) insertEventInTopology( func (s *outputRoomEventsTopologyStatements) InsertEventInTopology(
ctx context.Context, txn *sql.Tx, event *gomatrixserverlib.HeaderedEvent, pos types.StreamPosition, ctx context.Context, txn *sql.Tx, event *gomatrixserverlib.HeaderedEvent, pos types.StreamPosition,
) (err error) { ) (err error) {
stmt := common.TxStmt(txn, s.insertEventInTopologyStmt) stmt := common.TxStmt(txn, s.insertEventInTopologyStmt)
@ -118,7 +120,7 @@ func (s *outputRoomEventsTopologyStatements) insertEventInTopology(
// selectEventIDsInRange selects the IDs of events which positions are within a // selectEventIDsInRange selects the IDs of events which positions are within a
// given range in a given room's topological order. // given range in a given room's topological order.
// Returns an empty slice if no events match the given range. // Returns an empty slice if no events match the given range.
func (s *outputRoomEventsTopologyStatements) selectEventIDsInRange( func (s *outputRoomEventsTopologyStatements) SelectEventIDsInRange(
ctx context.Context, txn *sql.Tx, roomID string, ctx context.Context, txn *sql.Tx, roomID string,
fromPos, toPos, toMicroPos types.StreamPosition, fromPos, toPos, toMicroPos types.StreamPosition,
limit int, chronologicalOrder bool, limit int, chronologicalOrder bool,
@ -155,7 +157,7 @@ func (s *outputRoomEventsTopologyStatements) selectEventIDsInRange(
// selectPositionInTopology returns the position of a given event in the // selectPositionInTopology returns the position of a given event in the
// topology of the room it belongs to. // topology of the room it belongs to.
func (s *outputRoomEventsTopologyStatements) selectPositionInTopology( func (s *outputRoomEventsTopologyStatements) SelectPositionInTopology(
ctx context.Context, txn *sql.Tx, eventID string, ctx context.Context, txn *sql.Tx, eventID string,
) (pos types.StreamPosition, spos types.StreamPosition, err error) { ) (pos types.StreamPosition, spos types.StreamPosition, err error) {
stmt := common.TxStmt(txn, s.selectPositionInTopologyStmt) stmt := common.TxStmt(txn, s.selectPositionInTopologyStmt)
@ -163,7 +165,7 @@ func (s *outputRoomEventsTopologyStatements) selectPositionInTopology(
return return
} }
func (s *outputRoomEventsTopologyStatements) selectMaxPositionInTopology( func (s *outputRoomEventsTopologyStatements) SelectMaxPositionInTopology(
ctx context.Context, txn *sql.Tx, roomID string, ctx context.Context, txn *sql.Tx, roomID string,
) (pos types.StreamPosition, spos types.StreamPosition, err error) { ) (pos types.StreamPosition, spos types.StreamPosition, err error) {
stmt := common.TxStmt(txn, s.selectMaxPositionInTopologyStmt) stmt := common.TxStmt(txn, s.selectMaxPositionInTopologyStmt)
@ -173,7 +175,7 @@ func (s *outputRoomEventsTopologyStatements) selectMaxPositionInTopology(
// selectEventIDsFromPosition returns the IDs of all events that have a given // selectEventIDsFromPosition returns the IDs of all events that have a given
// position in the topology of a given room. // position in the topology of a given room.
func (s *outputRoomEventsTopologyStatements) selectEventIDsFromPosition( func (s *outputRoomEventsTopologyStatements) SelectEventIDsFromPosition(
ctx context.Context, txn *sql.Tx, roomID string, pos types.StreamPosition, ctx context.Context, txn *sql.Tx, roomID string, pos types.StreamPosition,
) (eventIDs []string, err error) { ) (eventIDs []string, err error) {
// Query the event IDs. // Query the event IDs.

1010
syncapi/storage/sqlite3/syncserver.go
View file

File diff suppressed because it is too large Load diff

2
syncapi/storage/storage.go
View file

@ -34,7 +34,7 @@ func NewSyncServerDatasource(dataSourceName string, dbProperties common.DbProper
case "postgres": case "postgres":
return postgres.NewDatabase(dataSourceName, dbProperties) return postgres.NewDatabase(dataSourceName, dbProperties)
case "file": case "file":
return sqlite3.NewSyncServerDatasource(dataSourceName) return sqlite3.NewDatabase(dataSourceName)
default: default:
return postgres.NewDatabase(dataSourceName, dbProperties) return postgres.NewDatabase(dataSourceName, dbProperties)
} }

12
syncapi/storage/storage_test.go
View file

@ -51,7 +51,7 @@ func MustCreateEvent(t *testing.T, roomID string, prevs []gomatrixserverlib.Head
} }
func MustCreateDatabase(t *testing.T) storage.Database { func MustCreateDatabase(t *testing.T) storage.Database {
db, err := sqlite3.NewSyncServerDatasource("file::memory:") db, err := sqlite3.NewDatabase("file::memory:")
if err != nil { if err != nil {
t.Fatalf("NewSyncServerDatasource returned %s", err) t.Fatalf("NewSyncServerDatasource returned %s", err)
} }
@ -222,7 +222,7 @@ func TestSyncResponse(t *testing.T) {
} }
func TestGetEventsInRangeWithPrevBatch(t *testing.T) { func TestGetEventsInRangeWithPrevBatch(t *testing.T) {
t.Parallel() //t.Parallel()
db := MustCreateDatabase(t) db := MustCreateDatabase(t)
events, _ := SimpleRoom(t, testRoomID, testUserIDA, testUserIDB) events, _ := SimpleRoom(t, testRoomID, testUserIDA, testUserIDB)
positions := MustWriteEvents(t, db, events) positions := MustWriteEvents(t, db, events)
@ -233,10 +233,10 @@ func TestGetEventsInRangeWithPrevBatch(t *testing.T) {
from := types.NewStreamToken( from := types.NewStreamToken(
positions[len(positions)-2], types.StreamPosition(0), positions[len(positions)-2], types.StreamPosition(0),
) )
db.IncrementalSync(ctx, testUserDeviceA, from, latest, 1, false)
res, err := db.IncrementalSync(ctx, testUserDeviceA, from, latest, 5, false) res, err := db.IncrementalSync(ctx, testUserDeviceA, from, latest, 5, false)
if err != nil { if err != nil {
t.Fatalf("failed to IncrementalSync with latest token") t.Fatalf("failed to IncrementalSync with latest token: %s", err)
} }
roomRes, ok := res.Rooms.Join[testRoomID] roomRes, ok := res.Rooms.Join[testRoomID]
if !ok { if !ok {
@ -258,7 +258,7 @@ func TestGetEventsInRangeWithPrevBatch(t *testing.T) {
to := types.NewTopologyToken(0, 0) to := types.NewTopologyToken(0, 0)
paginatedEvents, err := db.GetEventsInTopologicalRange(ctx, &prevBatchToken, &to, testRoomID, 5, true) paginatedEvents, err := db.GetEventsInTopologicalRange(ctx, &prevBatchToken, &to, testRoomID, 5, true)
if err != nil { if err != nil {
t.Fatalf("GetEventsInRange returned an error: %s", err) t.Fatalf("GetEventsInTopologicalRange returned an error: %s", err)
} }
gots := gomatrixserverlib.HeaderedToClientEvents(db.StreamEventsToEvents(&testUserDeviceA, paginatedEvents), gomatrixserverlib.FormatAll) gots := gomatrixserverlib.HeaderedToClientEvents(db.StreamEventsToEvents(&testUserDeviceA, paginatedEvents), gomatrixserverlib.FormatAll)
assertEventsEqual(t, "", true, gots, reversed(events[len(events)-6:len(events)-1])) assertEventsEqual(t, "", true, gots, reversed(events[len(events)-6:len(events)-1]))
@ -303,7 +303,7 @@ func TestGetEventsInRangeWithTopologyToken(t *testing.T) {
// backpaginate 5 messages starting at the latest position. // backpaginate 5 messages starting at the latest position.
paginatedEvents, err := db.GetEventsInTopologicalRange(ctx, &from, &to, testRoomID, 5, true) paginatedEvents, err := db.GetEventsInTopologicalRange(ctx, &from, &to, testRoomID, 5, true)
if err != nil { if err != nil {
t.Fatalf("GetEventsInRange returned an error: %s", err) t.Fatalf("GetEventsInTopologicalRange returned an error: %s", err)
} }
gots := gomatrixserverlib.HeaderedToClientEvents(db.StreamEventsToEvents(&testUserDeviceA, paginatedEvents), gomatrixserverlib.FormatAll) gots := gomatrixserverlib.HeaderedToClientEvents(db.StreamEventsToEvents(&testUserDeviceA, paginatedEvents), gomatrixserverlib.FormatAll)
assertEventsEqual(t, "", true, gots, reversed(events[len(events)-5:])) assertEventsEqual(t, "", true, gots, reversed(events[len(events)-5:]))

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

@ -34,18 +34,18 @@ type Events interface {
type Topology interface { type Topology interface {
// InsertEventInTopology inserts the given event in the room's topology, based // InsertEventInTopology inserts the given event in the room's topology, based
// on the event's depth. // on the event's depth.
InsertEventInTopology(ctx context.Context, event *gomatrixserverlib.HeaderedEvent, pos types.StreamPosition) (err error) InsertEventInTopology(ctx context.Context, txn *sql.Tx, event *gomatrixserverlib.HeaderedEvent, pos types.StreamPosition) (err error)
// SelectEventIDsInRange selects the IDs of events which positions are within a // SelectEventIDsInRange selects the IDs of events which positions are within a
// given range in a given room's topological order. // given range in a given room's topological order.
// Returns an empty slice if no events match the given range. // Returns an empty slice if no events match the given range.
SelectEventIDsInRange(ctx context.Context, roomID string, fromPos, toPos, toMicroPos types.StreamPosition, limit int, chronologicalOrder bool) (eventIDs []string, err error) SelectEventIDsInRange(ctx context.Context, txn *sql.Tx, roomID string, fromPos, toPos, toMicroPos types.StreamPosition, limit int, chronologicalOrder bool) (eventIDs []string, err error)
// SelectPositionInTopology returns the position of a given event in the // SelectPositionInTopology returns the position of a given event in the
// topology of the room it belongs to. // topology of the room it belongs to.
SelectPositionInTopology(ctx context.Context, eventID string) (pos, spos types.StreamPosition, err error) SelectPositionInTopology(ctx context.Context, txn *sql.Tx, eventID string) (pos, spos types.StreamPosition, err error)
SelectMaxPositionInTopology(ctx context.Context, roomID string) (pos types.StreamPosition, spos types.StreamPosition, err error) SelectMaxPositionInTopology(ctx context.Context, txn *sql.Tx, roomID string) (pos types.StreamPosition, spos types.StreamPosition, err error)
// SelectEventIDsFromPosition returns the IDs of all events that have a given // SelectEventIDsFromPosition returns the IDs of all events that have a given
// position in the topology of a given room. // position in the topology of a given room.
SelectEventIDsFromPosition(ctx context.Context, roomID string, pos types.StreamPosition) (eventIDs []string, err error) SelectEventIDsFromPosition(ctx context.Context, txn *sql.Tx, roomID string, pos types.StreamPosition) (eventIDs []string, err error)
} }
type CurrentRoomState interface { type CurrentRoomState interface {