Use TransactionWriter in other component SQLite (#1209)

* Use TransactionWriter on other component SQLites

* Fix sync API tests

* Fix panic in media API

* Fix a couple of transactions

* Fix wrong query, add some logging output

* Add debug logging into StoreEvent

* Adjust InsertRoomNID

* Update logging

6 files changed, 40 insertions, 24 deletions

diff --git a/roomserver/internal/input_events.go b/roomserver/internal/input_events.go
index 04538cf6..a6308299 100644
--- a/roomserver/internal/input_events.go
+++ b/roomserver/internal/input_events.go
@@ -18,6 +18,7 @@ package internal
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/matrix-org/dendrite/internal/eventutil"
 	"github.com/matrix-org/dendrite/roomserver/api"
@@ -65,13 +66,13 @@ func (r *RoomserverInternalAPI) processRoomEvent(
 	// Store the event.
 	roomNID, stateAtEvent, redactionEvent, redactedEventID, err := r.DB.StoreEvent(ctx, event, input.TransactionID, authEventNIDs)
 	if err != nil {
-		return
+		return "", fmt.Errorf("r.DB.StoreEvent: %w", err)
 	}
 	// if storing this event results in it being redacted then do so.
 	if redactedEventID == event.EventID() {
 		r, rerr := eventutil.RedactEvent(redactionEvent, &event)
 		if rerr != nil {
-			return "", rerr
+			return "", fmt.Errorf("eventutil.RedactEvent: %w", rerr)
 		}
 		event = *r
 	}
@@ -93,7 +94,7 @@ func (r *RoomserverInternalAPI) processRoomEvent(
 		// Lets calculate one.
 		err = r.calculateAndSetState(ctx, input, roomNID, &stateAtEvent, event)
 		if err != nil {
-			return
+			return "", fmt.Errorf("r.calculateAndSetState: %w", err)
 		}
 	}
 
@@ -105,7 +106,7 @@ func (r *RoomserverInternalAPI) processRoomEvent(
 		input.SendAsServer,  // send as server
 		input.TransactionID, // transaction ID
 	); err != nil {
-		return
+		return "", fmt.Errorf("r.updateLatestEvents: %w", err)
 	}
 
 	// processing this event resulted in an event (which may not be the one we're processing)
@@ -123,7 +124,7 @@ func (r *RoomserverInternalAPI) processRoomEvent(
 			},
 		})
 		if err != nil {
-			return
+			return "", fmt.Errorf("r.WriteOutputEvents: %w", err)
 		}
 	}
 
diff --git a/roomserver/storage/shared/storage.go b/roomserver/storage/shared/storage.go
index e2e5daf9..e858a9b0 100644
--- a/roomserver/storage/shared/storage.go
+++ b/roomserver/storage/shared/storage.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"encoding/json"
+	"fmt"
 
 	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/roomserver/api"
@@ -362,7 +363,7 @@ func (d *Database) StoreEvent(
 				ctx, txn, txnAndSessionID.TransactionID,
 				txnAndSessionID.SessionID, event.Sender(), event.EventID(),
 			); err != nil {
-				return err
+				return fmt.Errorf("d.TransactionsTable.InsertTransaction: %w", err)
 			}
 		}
 
@@ -377,15 +378,15 @@ func (d *Database) StoreEvent(
 		// room.
 		var roomVersion gomatrixserverlib.RoomVersion
 		if roomVersion, err = extractRoomVersionFromCreateEvent(event); err != nil {
-			return err
+			return fmt.Errorf("extractRoomVersionFromCreateEvent: %w", err)
 		}
 
 		if roomNID, err = d.assignRoomNID(ctx, txn, event.RoomID(), roomVersion); err != nil {
-			return err
+			return fmt.Errorf("d.assignRoomNID: %w", err)
 		}
 
 		if eventTypeNID, err = d.assignEventTypeNID(ctx, txn, event.Type()); err != nil {
-			return err
+			return fmt.Errorf("d.assignEventTypeNID: %w", err)
 		}
 
 		eventStateKey := event.StateKey()
@@ -393,7 +394,7 @@ func (d *Database) StoreEvent(
 		// Otherwise set the numeric ID for the state_key to 0.
 		if eventStateKey != nil {
 			if eventStateKeyNID, err = d.assignStateKeyNID(ctx, txn, *eventStateKey); err != nil {
-				return err
+				return fmt.Errorf("d.assignStateKeyNID: %w", err)
 			}
 		}
 
@@ -411,17 +412,20 @@ func (d *Database) StoreEvent(
 			if err == sql.ErrNoRows {
 				// We've already inserted the event so select the numeric event ID
 				eventNID, stateNID, err = d.EventsTable.SelectEvent(ctx, txn, event.EventID())
+				if err != nil {
+					return fmt.Errorf("d.EventsTable.SelectEvent: %w", err)
+				}
 			}
 			if err != nil {
-				return err
+				return fmt.Errorf("d.EventsTable.InsertEvent: %w", err)
 			}
 		}
 
 		if err = d.EventJSONTable.InsertEventJSON(ctx, txn, eventNID, event.JSON()); err != nil {
-			return err
+			return fmt.Errorf("d.EventJSONTable.InsertEventJSON: %w", err)
 		}
 		redactionEvent, redactedEventID, err = d.handleRedactions(ctx, txn, eventNID, event)
-		return err
+		return nil
 	})
 	if err != nil {
 		return 0, types.StateAtEvent{}, nil, "", err
diff --git a/roomserver/storage/sqlite3/events_table.go b/roomserver/storage/sqlite3/events_table.go
index 378441c3..b3cfee07 100644
--- a/roomserver/storage/sqlite3/events_table.go
+++ b/roomserver/storage/sqlite3/events_table.go
@@ -287,7 +287,8 @@ func (s *eventStatements) UpdateEventState(
 	ctx context.Context, eventNID types.EventNID, stateNID types.StateSnapshotNID,
 ) error {
 	return s.writer.Do(s.db, nil, func(txn *sql.Tx) error {
-		_, err := s.updateEventStateStmt.ExecContext(ctx, int64(stateNID), int64(eventNID))
+		stmt := sqlutil.TxStmt(txn, s.updateEventStateStmt)
+		_, err := stmt.ExecContext(ctx, int64(stateNID), int64(eventNID))
 		return err
 	})
 }
diff --git a/roomserver/storage/sqlite3/published_table.go b/roomserver/storage/sqlite3/published_table.go
index 96575241..85f1e0a4 100644
--- a/roomserver/storage/sqlite3/published_table.go
+++ b/roomserver/storage/sqlite3/published_table.go
@@ -71,7 +71,8 @@ func (s *publishedStatements) UpsertRoomPublished(
 	ctx context.Context, roomID string, published bool,
 ) (err error) {
 	return s.writer.Do(s.db, nil, func(txn *sql.Tx) error {
-		_, err := s.upsertPublishedStmt.ExecContext(ctx, roomID, published)
+		stmt := sqlutil.TxStmt(txn, s.upsertPublishedStmt)
+		_, err := stmt.ExecContext(ctx, roomID, published)
 		return err
 	})
 }
diff --git a/roomserver/storage/sqlite3/room_aliases_table.go b/roomserver/storage/sqlite3/room_aliases_table.go
index 096b73f9..4a535777 100644
--- a/roomserver/storage/sqlite3/room_aliases_table.go
+++ b/roomserver/storage/sqlite3/room_aliases_table.go
@@ -87,7 +87,8 @@ func (s *roomAliasesStatements) InsertRoomAlias(
 	ctx context.Context, alias string, roomID string, creatorUserID string,
 ) (err error) {
 	return s.writer.Do(s.db, nil, func(txn *sql.Tx) error {
-		_, err := s.insertRoomAliasStmt.ExecContext(ctx, alias, roomID, creatorUserID)
+		stmt := sqlutil.TxStmt(txn, s.insertRoomAliasStmt)
+		_, err := stmt.ExecContext(ctx, alias, roomID, creatorUserID)
 		return err
 	})
 }
@@ -139,7 +140,8 @@ func (s *roomAliasesStatements) DeleteRoomAlias(
 	ctx context.Context, alias string,
 ) (err error) {
 	return s.writer.Do(s.db, nil, func(txn *sql.Tx) error {
-		_, err := s.deleteRoomAliasStmt.ExecContext(ctx, alias)
+		stmt := sqlutil.TxStmt(txn, s.deleteRoomAliasStmt)
+		_, err := stmt.ExecContext(ctx, alias)
 		return err
 	})
 }
diff --git a/roomserver/storage/sqlite3/rooms_table.go b/roomserver/storage/sqlite3/rooms_table.go
index 9eeadea9..bb30a63b 100644
--- a/roomserver/storage/sqlite3/rooms_table.go
+++ b/roomserver/storage/sqlite3/rooms_table.go
@@ -20,6 +20,7 @@ import (
 	"database/sql"
 	"encoding/json"
 	"errors"
+	"fmt"
 
 	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/roomserver/storage/shared"
@@ -98,17 +99,23 @@ func NewSqliteRoomsTable(db *sql.DB) (tables.Rooms, error) {
 func (s *roomStatements) InsertRoomNID(
 	ctx context.Context, txn *sql.Tx,
 	roomID string, roomVersion gomatrixserverlib.RoomVersion,
-) (types.RoomNID, error) {
-	err := s.writer.Do(s.db, txn, func(txn *sql.Tx) error {
+) (roomNID types.RoomNID, err error) {
+	err = s.writer.Do(s.db, txn, func(txn *sql.Tx) error {
 		insertStmt := sqlutil.TxStmt(txn, s.insertRoomNIDStmt)
-		_, err := insertStmt.ExecContext(ctx, roomID, roomVersion)
-		return err
+		_, err = insertStmt.ExecContext(ctx, roomID, roomVersion)
+		if err != nil {
+			return fmt.Errorf("insertStmt.ExecContext: %w", err)
+		}
+		roomNID, err = s.SelectRoomNID(ctx, txn, roomID)
+		if err != nil {
+			return fmt.Errorf("s.SelectRoomNID: %w", err)
+		}
+		return nil
 	})
-	if err == nil {
-		return s.SelectRoomNID(ctx, txn, roomID)
-	} else {
+	if err != nil {
 		return types.RoomNID(0), err
 	}
+	return
 }
 
 func (s *roomStatements) SelectRoomNID(