Add sqlite migration

This commit is contained in:
Till Faelligen 2020-10-29 10:13:57 +01:00
parent 61f8d19c5c
commit 5fc9218c41
3 changed files with 158 additions and 63 deletions

View file

@ -0,0 +1,83 @@
// Copyright 2020 The Matrix.org Foundation C.I.C.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package deltas
import (
"database/sql"
"fmt"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/pressly/goose"
)
func LoadFromGoose() {
goose.AddMigration(UpAddForgottenColumn, DownAddForgottenColumn)
}
func LoadAddForgottenColumn(m *sqlutil.Migrations) {
m.AddMigration(UpAddForgottenColumn, DownAddForgottenColumn)
}
func UpAddForgottenColumn(tx *sql.Tx) error {
_, err := tx.Exec(` ALTER TABLE roomserver_membership RENAME TO roomserver_membership_tmp;
CREATE TABLE IF NOT EXISTS roomserver_membership (
room_nid INTEGER NOT NULL,
target_nid INTEGER NOT NULL,
sender_nid INTEGER NOT NULL DEFAULT 0,
membership_nid INTEGER NOT NULL DEFAULT 1,
event_nid INTEGER NOT NULL DEFAULT 0,
target_local BOOLEAN NOT NULL DEFAULT false,
forgotten BOOLEAN NOT NULL DEFAULT false,
UNIQUE (room_nid, target_nid)
);
INSERT
INTO roomserver_membership (
room_nid, target_nid, sender_nid, membership_nid, event_nid, target_local
) SELECT
room_nid, target_nid, sender_nid, membership_nid, event_nid, target_local
FROM roomserver_membership_tmp
;
DROP TABLE roomserver_membership_tmp;`)
if err != nil {
return fmt.Errorf("failed to execute upgrade: %w", err)
}
return nil
}
func DownAddForgottenColumn(tx *sql.Tx) error {
_, err := tx.Exec(` ALTER TABLE roomserver_membership RENAME TO roomserver_membership_tmp;
CREATE TABLE IF NOT EXISTS roomserver_membership (
room_nid INTEGER NOT NULL,
target_nid INTEGER NOT NULL,
sender_nid INTEGER NOT NULL DEFAULT 0,
membership_nid INTEGER NOT NULL DEFAULT 1,
event_nid INTEGER NOT NULL DEFAULT 0,
target_local BOOLEAN NOT NULL DEFAULT false,
UNIQUE (room_nid, target_nid)
);
INSERT
INTO roomserver_membership (
room_nid, target_nid, sender_nid, membership_nid, event_nid, target_local
) SELECT
room_nid, target_nid, sender_nid, membership_nid, event_nid, target_local
FROM roomserver_membership_tmp
;
DROP TABLE roomserver_membership_tmp;`)
if err != nil {
return fmt.Errorf("failed to execute downgrade: %w", err)
}
return nil
}

View file

@ -119,10 +119,6 @@ func NewSqliteMembershipTable(db *sql.DB) (tables.Membership, error) {
s := &membershipStatements{ s := &membershipStatements{
db: db, db: db,
} }
_, err := db.Exec(membershipSchema)
if err != nil {
return nil, err
}
return s, shared.StatementList{ return s, shared.StatementList{
{&s.insertMembershipStmt, insertMembershipSQL}, {&s.insertMembershipStmt, insertMembershipSQL},
@ -139,6 +135,11 @@ func NewSqliteMembershipTable(db *sql.DB) (tables.Membership, error) {
}.Prepare(db) }.Prepare(db)
} }
func (s *membershipStatements) execSchema(db *sql.DB) error {
_, err := db.Exec(membershipSchema)
return err
}
func (s *membershipStatements) InsertMembership( func (s *membershipStatements) InsertMembership(
ctx context.Context, txn *sql.Tx, ctx context.Context, txn *sql.Tx,
roomNID types.RoomNID, targetUserNID types.EventStateKeyNID, roomNID types.RoomNID, targetUserNID types.EventStateKeyNID,

View file

@ -19,127 +19,138 @@ import (
"context" "context"
"database/sql" "database/sql"
_ "github.com/mattn/go-sqlite3"
"github.com/matrix-org/dendrite/internal/caching" "github.com/matrix-org/dendrite/internal/caching"
"github.com/matrix-org/dendrite/internal/config" "github.com/matrix-org/dendrite/internal/config"
"github.com/matrix-org/dendrite/internal/sqlutil" "github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver/storage/shared" "github.com/matrix-org/dendrite/roomserver/storage/shared"
"github.com/matrix-org/dendrite/roomserver/storage/tables" "github.com/matrix-org/dendrite/roomserver/storage/sqlite3/deltas"
"github.com/matrix-org/dendrite/roomserver/types" "github.com/matrix-org/dendrite/roomserver/types"
"github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/gomatrixserverlib"
_ "github.com/mattn/go-sqlite3"
) )
// A Database is used to store room events and stream offsets. // A Database is used to store room events and stream offsets.
type Database struct { type Database struct {
shared.Database shared.Database
events tables.Events
eventJSON tables.EventJSON
eventTypes tables.EventTypes
eventStateKeys tables.EventStateKeys
rooms tables.Rooms
transactions tables.Transactions
prevEvents tables.PreviousEvents
invites tables.Invites
membership tables.Membership
db *sql.DB
writer sqlutil.Writer
} }
// Open a sqlite database. // Open a sqlite database.
// nolint: gocyclo
func Open(dbProperties *config.DatabaseOptions, cache caching.RoomServerCaches) (*Database, error) { func Open(dbProperties *config.DatabaseOptions, cache caching.RoomServerCaches) (*Database, error) {
var d Database var d Database
var db *sql.DB
var err error var err error
if d.db, err = sqlutil.Open(dbProperties); err != nil { if db, err = sqlutil.Open(dbProperties); err != nil {
return nil, err return nil, err
} }
d.writer = sqlutil.NewExclusiveWriter()
//d.db.Exec("PRAGMA journal_mode=WAL;") //db.Exec("PRAGMA journal_mode=WAL;")
//d.db.Exec("PRAGMA read_uncommitted = true;") //db.Exec("PRAGMA read_uncommitted = true;")
// FIXME: We are leaking connections somewhere. Setting this to 2 will eventually // FIXME: We are leaking connections somewhere. Setting this to 2 will eventually
// cause the roomserver to be unresponsive to new events because something will // cause the roomserver to be unresponsive to new events because something will
// acquire the global mutex and never unlock it because it is waiting for a connection // acquire the global mutex and never unlock it because it is waiting for a connection
// which it will never obtain. // which it will never obtain.
d.db.SetMaxOpenConns(20) db.SetMaxOpenConns(20)
d.eventStateKeys, err = NewSqliteEventStateKeysTable(d.db) // Create tables before executing migrations so we don't fail if the table is missing,
if err != nil { // and THEN prepare statements so we don't fail due to referencing new columns
ms := membershipStatements{}
if err := ms.execSchema(db); err != nil {
return nil, err return nil, err
} }
d.eventTypes, err = NewSqliteEventTypesTable(d.db) m := sqlutil.NewMigrations()
if err != nil { deltas.LoadAddForgottenColumn(m)
if err := m.RunDeltas(db, dbProperties); err != nil {
return nil, err return nil, err
} }
d.eventJSON, err = NewSqliteEventJSONTable(d.db) if err := d.prepare(db, cache); err != nil {
if err != nil {
return nil, err return nil, err
} }
d.events, err = NewSqliteEventsTable(d.db)
return &d, nil
}
// nolint: gocyclo
func (d *Database) prepare(db *sql.DB, cache caching.RoomServerCaches) error {
var err error
eventStateKeys, err := NewSqliteEventStateKeysTable(db)
if err != nil { if err != nil {
return nil, err return err
} }
d.rooms, err = NewSqliteRoomsTable(d.db) eventTypes, err := NewSqliteEventTypesTable(db)
if err != nil { if err != nil {
return nil, err return err
} }
d.transactions, err = NewSqliteTransactionsTable(d.db) eventJSON, err := NewSqliteEventJSONTable(db)
if err != nil { if err != nil {
return nil, err return err
} }
stateBlock, err := NewSqliteStateBlockTable(d.db) events, err := NewSqliteEventsTable(db)
if err != nil { if err != nil {
return nil, err return err
} }
stateSnapshot, err := NewSqliteStateSnapshotTable(d.db) rooms, err := NewSqliteRoomsTable(db)
if err != nil { if err != nil {
return nil, err return err
} }
d.prevEvents, err = NewSqlitePrevEventsTable(d.db) transactions, err := NewSqliteTransactionsTable(db)
if err != nil { if err != nil {
return nil, err return err
} }
roomAliases, err := NewSqliteRoomAliasesTable(d.db) stateBlock, err := NewSqliteStateBlockTable(db)
if err != nil { if err != nil {
return nil, err return err
} }
d.invites, err = NewSqliteInvitesTable(d.db) stateSnapshot, err := NewSqliteStateSnapshotTable(db)
if err != nil { if err != nil {
return nil, err return err
} }
d.membership, err = NewSqliteMembershipTable(d.db) prevEvents, err := NewSqlitePrevEventsTable(db)
if err != nil { if err != nil {
return nil, err return err
} }
published, err := NewSqlitePublishedTable(d.db) roomAliases, err := NewSqliteRoomAliasesTable(db)
if err != nil { if err != nil {
return nil, err return err
} }
redactions, err := NewSqliteRedactionsTable(d.db) invites, err := NewSqliteInvitesTable(db)
if err != nil { if err != nil {
return nil, err return err
}
membership, err := NewSqliteMembershipTable(db)
if err != nil {
return err
}
published, err := NewSqlitePublishedTable(db)
if err != nil {
return err
}
redactions, err := NewSqliteRedactionsTable(db)
if err != nil {
return err
} }
d.Database = shared.Database{ d.Database = shared.Database{
DB: d.db, DB: db,
Cache: cache, Cache: cache,
Writer: d.writer, Writer: sqlutil.NewExclusiveWriter(),
EventsTable: d.events, EventsTable: events,
EventTypesTable: d.eventTypes, EventTypesTable: eventTypes,
EventStateKeysTable: d.eventStateKeys, EventStateKeysTable: eventStateKeys,
EventJSONTable: d.eventJSON, EventJSONTable: eventJSON,
RoomsTable: d.rooms, RoomsTable: rooms,
TransactionsTable: d.transactions, TransactionsTable: transactions,
StateBlockTable: stateBlock, StateBlockTable: stateBlock,
StateSnapshotTable: stateSnapshot, StateSnapshotTable: stateSnapshot,
PrevEventsTable: d.prevEvents, PrevEventsTable: prevEvents,
RoomAliasesTable: roomAliases, RoomAliasesTable: roomAliases,
InvitesTable: d.invites, InvitesTable: invites,
MembershipTable: d.membership, MembershipTable: membership,
PublishedTable: published, PublishedTable: published,
RedactionsTable: redactions, RedactionsTable: redactions,
GetLatestEventsForUpdateFn: d.GetLatestEventsForUpdate, GetLatestEventsForUpdateFn: d.GetLatestEventsForUpdate,
} }
return &d, nil return nil
} }
func (d *Database) SupportsConcurrentRoomInputs() bool { func (d *Database) SupportsConcurrentRoomInputs() bool {