Roomserver database-wide TransactionWriters (#1282)

* Database-wide TransactionWriter

* Fix deadlocking Sync API tests

* Undo non-roomserver changes for now

15 files changed, 44 insertions, 43 deletions

diff --git a/roomserver/storage/sqlite3/event_json_table.go b/roomserver/storage/sqlite3/event_json_table.go
index 64795d02..e8118ad7 100644
--- a/roomserver/storage/sqlite3/event_json_table.go
+++ b/roomserver/storage/sqlite3/event_json_table.go
@@ -54,10 +54,10 @@ type eventJSONStatements struct {
 	bulkSelectEventJSONStmt *sql.Stmt
 }
 
-func NewSqliteEventJSONTable(db *sql.DB) (tables.EventJSON, error) {
+func NewSqliteEventJSONTable(db *sql.DB, writer *sqlutil.TransactionWriter) (tables.EventJSON, error) {
 	s := &eventJSONStatements{
 		db:     db,
-		writer: sqlutil.NewTransactionWriter(),
+		writer: writer,
 	}
 	_, err := db.Exec(eventJSONSchema)
 	if err != nil {
diff --git a/roomserver/storage/sqlite3/event_state_keys_table.go b/roomserver/storage/sqlite3/event_state_keys_table.go
index 3e9f2e61..c8ad052b 100644
--- a/roomserver/storage/sqlite3/event_state_keys_table.go
+++ b/roomserver/storage/sqlite3/event_state_keys_table.go
@@ -71,10 +71,10 @@ type eventStateKeyStatements struct {
 	bulkSelectEventStateKeyStmt    *sql.Stmt
 }
 
-func NewSqliteEventStateKeysTable(db *sql.DB) (tables.EventStateKeys, error) {
+func NewSqliteEventStateKeysTable(db *sql.DB, writer *sqlutil.TransactionWriter) (tables.EventStateKeys, error) {
 	s := &eventStateKeyStatements{
 		db:     db,
-		writer: sqlutil.NewTransactionWriter(),
+		writer: writer,
 	}
 	_, err := db.Exec(eventStateKeysSchema)
 	if err != nil {
diff --git a/roomserver/storage/sqlite3/event_types_table.go b/roomserver/storage/sqlite3/event_types_table.go
index fd4a2e42..4a645789 100644
--- a/roomserver/storage/sqlite3/event_types_table.go
+++ b/roomserver/storage/sqlite3/event_types_table.go
@@ -85,10 +85,10 @@ type eventTypeStatements struct {
 	bulkSelectEventTypeNIDStmt   *sql.Stmt
 }
 
-func NewSqliteEventTypesTable(db *sql.DB) (tables.EventTypes, error) {
+func NewSqliteEventTypesTable(db *sql.DB, writer *sqlutil.TransactionWriter) (tables.EventTypes, error) {
 	s := &eventTypeStatements{
 		db:     db,
-		writer: sqlutil.NewTransactionWriter(),
+		writer: writer,
 	}
 	_, err := db.Exec(eventTypesSchema)
 	if err != nil {
diff --git a/roomserver/storage/sqlite3/events_table.go b/roomserver/storage/sqlite3/events_table.go
index b3cfee07..3ac30ca3 100644
--- a/roomserver/storage/sqlite3/events_table.go
+++ b/roomserver/storage/sqlite3/events_table.go
@@ -115,10 +115,10 @@ type eventStatements struct {
 	selectRoomNIDForEventNIDStmt           *sql.Stmt
 }
 
-func NewSqliteEventsTable(db *sql.DB) (tables.Events, error) {
+func NewSqliteEventsTable(db *sql.DB, writer *sqlutil.TransactionWriter) (tables.Events, error) {
 	s := &eventStatements{
 		db:     db,
-		writer: sqlutil.NewTransactionWriter(),
+		writer: writer,
 	}
 	_, err := db.Exec(eventsSchema)
 	if err != nil {
diff --git a/roomserver/storage/sqlite3/invite_table.go b/roomserver/storage/sqlite3/invite_table.go
index e806eab6..1305f4a8 100644
--- a/roomserver/storage/sqlite3/invite_table.go
+++ b/roomserver/storage/sqlite3/invite_table.go
@@ -71,10 +71,10 @@ type inviteStatements struct {
 	selectInvitesAboutToRetireStmt      *sql.Stmt
 }
 
-func NewSqliteInvitesTable(db *sql.DB) (tables.Invites, error) {
+func NewSqliteInvitesTable(db *sql.DB, writer *sqlutil.TransactionWriter) (tables.Invites, error) {
 	s := &inviteStatements{
 		db:     db,
-		writer: sqlutil.NewTransactionWriter(),
+		writer: writer,
 	}
 	_, err := db.Exec(inviteSchema)
 	if err != nil {
@@ -124,7 +124,7 @@ func (s *inviteStatements) UpdateInviteRetired(
 		if err != nil {
 			return err
 		}
-		defer (func() { err = rows.Close() })()
+		defer internal.CloseAndLogIfError(ctx, rows, "UpdateInviteRetired: rows.close() failed")
 		for rows.Next() {
 			var inviteEventID string
 			if err = rows.Scan(&inviteEventID); err != nil {
diff --git a/roomserver/storage/sqlite3/membership_table.go b/roomserver/storage/sqlite3/membership_table.go
index 6dd8bd83..7b69cee3 100644
--- a/roomserver/storage/sqlite3/membership_table.go
+++ b/roomserver/storage/sqlite3/membership_table.go
@@ -88,10 +88,10 @@ type membershipStatements struct {
 	updateMembershipStmt                            *sql.Stmt
 }
 
-func NewSqliteMembershipTable(db *sql.DB) (tables.Membership, error) {
+func NewSqliteMembershipTable(db *sql.DB, writer *sqlutil.TransactionWriter) (tables.Membership, error) {
 	s := &membershipStatements{
 		db:     db,
-		writer: sqlutil.NewTransactionWriter(),
+		writer: writer,
 	}
 	_, err := db.Exec(membershipSchema)
 	if err != nil {
diff --git a/roomserver/storage/sqlite3/previous_events_table.go b/roomserver/storage/sqlite3/previous_events_table.go
index 28b5d18f..ff804861 100644
--- a/roomserver/storage/sqlite3/previous_events_table.go
+++ b/roomserver/storage/sqlite3/previous_events_table.go
@@ -59,10 +59,10 @@ type previousEventStatements struct {
 	selectPreviousEventExistsStmt *sql.Stmt
 }
 
-func NewSqlitePrevEventsTable(db *sql.DB) (tables.PreviousEvents, error) {
+func NewSqlitePrevEventsTable(db *sql.DB, writer *sqlutil.TransactionWriter) (tables.PreviousEvents, error) {
 	s := &previousEventStatements{
 		db:     db,
-		writer: sqlutil.NewTransactionWriter(),
+		writer: writer,
 	}
 	_, err := db.Exec(previousEventSchema)
 	if err != nil {
diff --git a/roomserver/storage/sqlite3/published_table.go b/roomserver/storage/sqlite3/published_table.go
index 85f1e0a4..a4a47aec 100644
--- a/roomserver/storage/sqlite3/published_table.go
+++ b/roomserver/storage/sqlite3/published_table.go
@@ -51,10 +51,10 @@ type publishedStatements struct {
 	selectPublishedStmt    *sql.Stmt
 }
 
-func NewSqlitePublishedTable(db *sql.DB) (tables.Published, error) {
+func NewSqlitePublishedTable(db *sql.DB, writer *sqlutil.TransactionWriter) (tables.Published, error) {
 	s := &publishedStatements{
 		db:     db,
-		writer: sqlutil.NewTransactionWriter(),
+		writer: writer,
 	}
 	_, err := db.Exec(publishedSchema)
 	if err != nil {
diff --git a/roomserver/storage/sqlite3/redactions_table.go b/roomserver/storage/sqlite3/redactions_table.go
index d2bd2a20..ad900a4e 100644
--- a/roomserver/storage/sqlite3/redactions_table.go
+++ b/roomserver/storage/sqlite3/redactions_table.go
@@ -60,10 +60,10 @@ type redactionStatements struct {
 	markRedactionValidatedStmt                  *sql.Stmt
 }
 
-func NewSqliteRedactionsTable(db *sql.DB) (tables.Redactions, error) {
+func NewSqliteRedactionsTable(db *sql.DB, writer *sqlutil.TransactionWriter) (tables.Redactions, error) {
 	s := &redactionStatements{
 		db:     db,
-		writer: sqlutil.NewTransactionWriter(),
+		writer: writer,
 	}
 	_, err := db.Exec(redactionsSchema)
 	if err != nil {
diff --git a/roomserver/storage/sqlite3/room_aliases_table.go b/roomserver/storage/sqlite3/room_aliases_table.go
index 4a535777..deba3ff5 100644
--- a/roomserver/storage/sqlite3/room_aliases_table.go
+++ b/roomserver/storage/sqlite3/room_aliases_table.go
@@ -65,10 +65,10 @@ type roomAliasesStatements struct {
 	deleteRoomAliasStmt          *sql.Stmt
 }
 
-func NewSqliteRoomAliasesTable(db *sql.DB) (tables.RoomAliases, error) {
+func NewSqliteRoomAliasesTable(db *sql.DB, writer *sqlutil.TransactionWriter) (tables.RoomAliases, error) {
 	s := &roomAliasesStatements{
 		db:     db,
-		writer: sqlutil.NewTransactionWriter(),
+		writer: writer,
 	}
 	_, err := db.Exec(roomAliasesSchema)
 	if err != nil {
diff --git a/roomserver/storage/sqlite3/rooms_table.go b/roomserver/storage/sqlite3/rooms_table.go
index bb30a63b..8bbec508 100644
--- a/roomserver/storage/sqlite3/rooms_table.go
+++ b/roomserver/storage/sqlite3/rooms_table.go
@@ -76,10 +76,10 @@ type roomStatements struct {
 	selectRoomVersionForRoomNIDStmt    *sql.Stmt
 }
 
-func NewSqliteRoomsTable(db *sql.DB) (tables.Rooms, error) {
+func NewSqliteRoomsTable(db *sql.DB, writer *sqlutil.TransactionWriter) (tables.Rooms, error) {
 	s := &roomStatements{
 		db:     db,
-		writer: sqlutil.NewTransactionWriter(),
+		writer: writer,
 	}
 	_, err := db.Exec(roomsSchema)
 	if err != nil {
diff --git a/roomserver/storage/sqlite3/state_block_table.go b/roomserver/storage/sqlite3/state_block_table.go
index 3d716b64..3e28e450 100644
--- a/roomserver/storage/sqlite3/state_block_table.go
+++ b/roomserver/storage/sqlite3/state_block_table.go
@@ -81,10 +81,10 @@ type stateBlockStatements struct {
 	bulkSelectFilteredStateBlockEntriesStmt *sql.Stmt
 }
 
-func NewSqliteStateBlockTable(db *sql.DB) (tables.StateBlock, error) {
+func NewSqliteStateBlockTable(db *sql.DB, writer *sqlutil.TransactionWriter) (tables.StateBlock, error) {
 	s := &stateBlockStatements{
 		db:     db,
-		writer: sqlutil.NewTransactionWriter(),
+		writer: writer,
 	}
 	_, err := db.Exec(stateDataSchema)
 	if err != nil {
diff --git a/roomserver/storage/sqlite3/state_snapshot_table.go b/roomserver/storage/sqlite3/state_snapshot_table.go
index 48f1210b..799904ff 100644
--- a/roomserver/storage/sqlite3/state_snapshot_table.go
+++ b/roomserver/storage/sqlite3/state_snapshot_table.go
@@ -55,10 +55,10 @@ type stateSnapshotStatements struct {
 	bulkSelectStateBlockNIDsStmt *sql.Stmt
 }
 
-func NewSqliteStateSnapshotTable(db *sql.DB) (tables.StateSnapshot, error) {
+func NewSqliteStateSnapshotTable(db *sql.DB, writer *sqlutil.TransactionWriter) (tables.StateSnapshot, error) {
 	s := &stateSnapshotStatements{
 		db:     db,
-		writer: sqlutil.NewTransactionWriter(),
+		writer: writer,
 	}
 	_, err := db.Exec(stateSnapshotSchema)
 	if err != nil {
diff --git a/roomserver/storage/sqlite3/storage.go b/roomserver/storage/sqlite3/storage.go
index 048de192..ae3140d7 100644
--- a/roomserver/storage/sqlite3/storage.go
+++ b/roomserver/storage/sqlite3/storage.go
@@ -51,6 +51,7 @@ func Open(dbProperties *config.DatabaseOptions) (*Database, error) {
 	if d.db, err = sqlutil.Open(dbProperties); err != nil {
 		return nil, err
 	}
+	writer := sqlutil.NewTransactionWriter()
 	//d.db.Exec("PRAGMA journal_mode=WAL;")
 	//d.db.Exec("PRAGMA read_uncommitted = true;")
 
@@ -60,59 +61,59 @@ func Open(dbProperties *config.DatabaseOptions) (*Database, error) {
 	// which it will never obtain.
 	d.db.SetMaxOpenConns(20)
 
-	d.eventStateKeys, err = NewSqliteEventStateKeysTable(d.db)
+	d.eventStateKeys, err = NewSqliteEventStateKeysTable(d.db, writer)
 	if err != nil {
 		return nil, err
 	}
-	d.eventTypes, err = NewSqliteEventTypesTable(d.db)
+	d.eventTypes, err = NewSqliteEventTypesTable(d.db, writer)
 	if err != nil {
 		return nil, err
 	}
-	d.eventJSON, err = NewSqliteEventJSONTable(d.db)
+	d.eventJSON, err = NewSqliteEventJSONTable(d.db, writer)
 	if err != nil {
 		return nil, err
 	}
-	d.events, err = NewSqliteEventsTable(d.db)
+	d.events, err = NewSqliteEventsTable(d.db, writer)
 	if err != nil {
 		return nil, err
 	}
-	d.rooms, err = NewSqliteRoomsTable(d.db)
+	d.rooms, err = NewSqliteRoomsTable(d.db, writer)
 	if err != nil {
 		return nil, err
 	}
-	d.transactions, err = NewSqliteTransactionsTable(d.db)
+	d.transactions, err = NewSqliteTransactionsTable(d.db, writer)
 	if err != nil {
 		return nil, err
 	}
-	stateBlock, err := NewSqliteStateBlockTable(d.db)
+	stateBlock, err := NewSqliteStateBlockTable(d.db, writer)
 	if err != nil {
 		return nil, err
 	}
-	stateSnapshot, err := NewSqliteStateSnapshotTable(d.db)
+	stateSnapshot, err := NewSqliteStateSnapshotTable(d.db, writer)
 	if err != nil {
 		return nil, err
 	}
-	d.prevEvents, err = NewSqlitePrevEventsTable(d.db)
+	d.prevEvents, err = NewSqlitePrevEventsTable(d.db, writer)
 	if err != nil {
 		return nil, err
 	}
-	roomAliases, err := NewSqliteRoomAliasesTable(d.db)
+	roomAliases, err := NewSqliteRoomAliasesTable(d.db, writer)
 	if err != nil {
 		return nil, err
 	}
-	d.invites, err = NewSqliteInvitesTable(d.db)
+	d.invites, err = NewSqliteInvitesTable(d.db, writer)
 	if err != nil {
 		return nil, err
 	}
-	d.membership, err = NewSqliteMembershipTable(d.db)
+	d.membership, err = NewSqliteMembershipTable(d.db, writer)
 	if err != nil {
 		return nil, err
 	}
-	published, err := NewSqlitePublishedTable(d.db)
+	published, err := NewSqlitePublishedTable(d.db, writer)
 	if err != nil {
 		return nil, err
 	}
-	redactions, err := NewSqliteRedactionsTable(d.db)
+	redactions, err := NewSqliteRedactionsTable(d.db, writer)
 	if err != nil {
 		return nil, err
 	}
diff --git a/roomserver/storage/sqlite3/transactions_table.go b/roomserver/storage/sqlite3/transactions_table.go
index 2f6cff95..65c18a8a 100644
--- a/roomserver/storage/sqlite3/transactions_table.go
+++ b/roomserver/storage/sqlite3/transactions_table.go
@@ -50,10 +50,10 @@ type transactionStatements struct {
 	selectTransactionEventIDStmt *sql.Stmt
 }
 
-func NewSqliteTransactionsTable(db *sql.DB) (tables.Transactions, error) {
+func NewSqliteTransactionsTable(db *sql.DB, writer *sqlutil.TransactionWriter) (tables.Transactions, error) {
 	s := &transactionStatements{
 		db:     db,
-		writer: sqlutil.NewTransactionWriter(),
+		writer: writer,
 	}
 	_, err := db.Exec(transactionsSchema)
 	if err != nil {