mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2026-01-11 16:13:10 -06:00
Code Issues Packages Projects Releases Wiki Activity

Drop column in roomserver_previous_events

Browse source
This commit is contained in:
Till Faelligen 2023-05-17 11:28:03 +02:00
parent fe5e81e6b1
commit ec5d23005e
No known key found for this signature in database
GPG key ID: 3DF82D8AB9211D4E
8 changed files with 101 additions and 22 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
roomserver/storage/postgres/deltas/20230516154000_drop_reference_sha.go
View file

@ -20,7 +20,7 @@ import (
"fmt"
)
func UpDropEventReferenceSHA(ctx context.Context, tx *sql.Tx) error {
func UpDropEventReferenceSHAEvents(ctx context.Context, tx *sql.Tx) error {
var count int
err := tx.QueryRowContext(ctx, `SELECT count(*) FROM roomserver_events GROUP BY event_id HAVING count(event_id) > 1`).
Scan(&count)
@ -36,3 +36,11 @@ func UpDropEventReferenceSHA(ctx context.Context, tx *sql.Tx) error {
}
return nil
}
func UpDropEventReferenceSHAPrevEvents(ctx context.Context, tx *sql.Tx) error {
_, err := tx.ExecContext(ctx, `ALTER TABLE roomserver_previous_events DROP COLUMN IF EXISTS previous_reference_sha256;`)
if err != nil {
return fmt.Errorf("failed to execute upgrade: %w", err)
}
return nil
}

2
roomserver/storage/postgres/events_table.go
View file

@ -184,7 +184,7 @@ func CreateEventsTable(db *sql.DB) error {
m.AddMigrations([]sqlutil.Migration{
{
Version: "roomserver: drop column reference_sha from roomserver_events",
Up: deltas.UpDropEventReferenceSHA,
Up: deltas.UpDropEventReferenceSHAEvents,
},
}...)
return m.Up(context.Background())

28
roomserver/storage/postgres/previous_events_table.go
View file

@ -20,6 +20,7 @@ import (
"database/sql"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver/storage/postgres/deltas"
"github.com/matrix-org/dendrite/roomserver/storage/tables"
"github.com/matrix-org/dendrite/roomserver/types"
)
@ -32,11 +33,9 @@ const previousEventSchema = `
CREATE TABLE IF NOT EXISTS roomserver_previous_events (
-- The string event ID taken from the prev_events key of an event.
previous_event_id TEXT NOT NULL,
-- The SHA256 reference hash taken from the prev_events key of an event.
previous_reference_sha256 BYTEA NOT NULL,
-- A list of numeric event IDs of events that reference this prev_event.
event_nids BIGINT[] NOT NULL,
CONSTRAINT roomserver_previous_event_id_unique UNIQUE (previous_event_id, previous_reference_sha256)
CONSTRAINT roomserver_previous_event_id_unique UNIQUE (previous_event_id)
);
`
@ -47,11 +46,11 @@ CREATE TABLE IF NOT EXISTS roomserver_previous_events (
// The lock is necessary to avoid data races when checking whether an event is already referenced by another event.
const insertPreviousEventSQL = "" +
"INSERT INTO roomserver_previous_events" +
" (previous_event_id, previous_reference_sha256, event_nids)" +
" VALUES ($1, $2, array_append('{}'::bigint[], $3))" +
" (previous_event_id, event_nids)" +
" VALUES ($1, array_append('{}'::bigint[], $2))" +
" ON CONFLICT ON CONSTRAINT roomserver_previous_event_id_unique" +
" DO UPDATE SET event_nids = array_append(roomserver_previous_events.event_nids, $3)" +
" WHERE $3 != ALL(roomserver_previous_events.event_nids)"
" DO UPDATE SET event_nids = array_append(roomserver_previous_events.event_nids, $2)" +
" WHERE $2 != ALL(roomserver_previous_events.event_nids)"
// Check if the event is referenced by another event in the table.
// This should only be done while holding a "FOR UPDATE" lock on the row in the rooms table for this room.
@ -66,7 +65,18 @@ type previousEventStatements struct {
func CreatePrevEventsTable(db *sql.DB) error {
_, err := db.Exec(previousEventSchema)
return err
if err != nil {
return err
}
m := sqlutil.NewMigrator(db)
m.AddMigrations([]sqlutil.Migration{
{
Version: "roomserver: drop column reference_sha from roomserver_prev_events",
Up: deltas.UpDropEventReferenceSHAPrevEvents,
},
}...)
return m.Up(context.Background())
}
func PreparePrevEventsTable(db *sql.DB) (tables.PreviousEvents, error) {
@ -86,7 +96,7 @@ func (s *previousEventStatements) InsertPreviousEvent(
) error {
stmt := sqlutil.TxStmt(txn, s.insertPreviousEventStmt)
_, err := stmt.ExecContext(
ctx, previousEventID, []byte(""), int64(eventNID),
ctx, previousEventID, int64(eventNID),
)
return err
}

4
roomserver/storage/shared/room_updater.go
View file

@ -191,8 +191,8 @@ func (u *RoomUpdater) EventsFromIDs(ctx context.Context, roomInfo *types.RoomInf
}
// IsReferenced implements types.RoomRecentEventsUpdater
func (u *RoomUpdater) IsReferenced(eventReference gomatrixserverlib.EventReference) (bool, error) {
err := u.d.PrevEventsTable.SelectPreviousEventExists(u.ctx, u.txn, eventReference.EventID)
func (u *RoomUpdater) IsReferenced(eventID string) (bool, error) {
err := u.d.PrevEventsTable.SelectPreviousEventExists(u.ctx, u.txn, eventID)
if err == nil {
return true, nil
}

34
roomserver/storage/sqlite3/deltas/20230516154000_drop_reference_sha.go
View file

@ -36,3 +36,37 @@ func UpDropEventReferenceSHA(ctx context.Context, tx *sql.Tx) error {
}
return nil
}
func UpDropEventReferenceSHAPrevEvents(ctx context.Context, tx *sql.Tx) error {
// rename the table
if _, err := tx.ExecContext(ctx, `ALTER TABLE roomserver_previous_events RENAME TO _roomserver_previous_events;`); err != nil {
return fmt.Errorf("tx.ExecContext: %w", err)
}
// create new table
if _, err := tx.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS roomserver_previous_events (
previous_event_id TEXT NOT NULL,
event_nids TEXT NOT NULL,
UNIQUE (previous_event_id)
);`); err != nil {
return fmt.Errorf("tx.ExecContext: %w", err)
}
// move data
if _, err := tx.ExecContext(ctx, `
INSERT
INTO roomserver_previous_events (
previous_event_id, event_nids
) SELECT
previous_event_id, event_nids
FROM _roomserver_previous_events
;`); err != nil {
return fmt.Errorf("tx.ExecContext: %w", err)
}
// drop old table
_, err := tx.ExecContext(ctx, `DROP TABLE _roomserver_previous_events;`)
if err != nil {
return fmt.Errorf("failed to execute upgrade: %w", err)
}
return nil
}

35
roomserver/storage/sqlite3/previous_events_table.go
View file

@ -18,12 +18,15 @@ package sqlite3
import (
"context"
"database/sql"
"errors"
"fmt"
"strings"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver/storage/sqlite3/deltas"
"github.com/matrix-org/dendrite/roomserver/storage/tables"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/sirupsen/logrus"
)
// TODO: previous_reference_sha256 was NOT NULL before but it broke sytest because
@ -34,9 +37,8 @@ import (
const previousEventSchema = `
CREATE TABLE IF NOT EXISTS roomserver_previous_events (
previous_event_id TEXT NOT NULL,
previous_reference_sha256 BLOB,
event_nids TEXT NOT NULL,
UNIQUE (previous_event_id, previous_reference_sha256)
UNIQUE (previous_event_id)
);
`
@ -72,7 +74,34 @@ type previousEventStatements struct {
func CreatePrevEventsTable(db *sql.DB) error {
_, err := db.Exec(previousEventSchema)
return err
if err != nil {
return err
}
// check if the column exists
var cName string
migrationName := "roomserver: drop column reference_sha from roomserver_prev_events"
err = db.QueryRowContext(context.Background(), `SELECT p.name FROM sqlite_master AS m JOIN pragma_table_info(m.name) AS p WHERE m.name = 'roomserver_previous_events' AND p.name = 'previous_reference_sha256'`).Scan(&cName)
if err != nil {
if errors.Is(err, sql.ErrNoRows) { // migration was already executed, as the column was removed
if err = sqlutil.InsertMigration(context.Background(), db, migrationName); err != nil {
return fmt.Errorf("unable to manually insert migration '%s': %w", migrationName, err)
}
return nil
}
return err
}
m := sqlutil.NewMigrator(db)
m.AddMigrations([]sqlutil.Migration{
{
Version: migrationName,
Up: deltas.UpDropEventReferenceSHAPrevEvents,
},
}...)
logrus.Infof("cName: %s", cName)
if err := m.Up(context.Background()); err != nil {
panic(err)
}
return nil
}
func PreparePrevEventsTable(db *sql.DB) (tables.PreviousEvents, error) {

3
roomserver/storage/sqlite3/storage.go
View file

@ -21,14 +21,13 @@ import (
"errors"
"fmt"
"github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver/storage/shared"
"github.com/matrix-org/dendrite/roomserver/storage/sqlite3/deltas"
"github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/gomatrixserverlib"
)
// A Database is used to store room events and stream offsets.

7
roomserver/storage/tables/events_table_test.go
View file

@ -75,8 +75,7 @@ func Test_EventsTable(t *testing.T) {
assert.True(t, sentToOutput)
eventIDs = append(eventIDs, ev.EventID())
ref := ev.EventReference()
ref.EventSHA256 = nil
ref := gomatrixserverlib.EventReference{EventID: eventID}
wantEventReferences = append(wantEventReferences, ref)
// Set the stateSnapshot to 2 for some events to verify they are returned later
@ -99,8 +98,8 @@ func Test_EventsTable(t *testing.T) {
}
wantStateAtEvent = append(wantStateAtEvent, stateAtEvent)
wantStateAtEventAndRefs = append(wantStateAtEventAndRefs, types.StateAtEventAndReference{
StateAtEvent: stateAtEvent,
EventReference: ref,
StateAtEvent: stateAtEvent,
EventID: ev.EventID(),
})
}