Merge branch 'main' into neilalexander/genericapis

keyserver/storage/postgres/key_changes_table.go

@@ -18,6 +18,8 @@ import (
 	"context"
 	"database/sql"
 
+	"github.com/lib/pq"
+
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/keyserver/storage/postgres/deltas"
@@ -64,7 +66,8 @@ func NewPostgresKeyChangesTable(db *sql.DB) (tables.KeyChanges, error) {
 	// TODO: Remove when we are sure we are not having goose artefacts in the db
 	// This forces an error, which indicates the migration is already applied, since the
 	// column partition was removed from the table
-	err = db.QueryRow("SELECT partition FROM keyserver_key_changes LIMIT 1;").Scan()
+	var count int
+	err = db.QueryRow("SELECT partition FROM keyserver_key_changes LIMIT 1;").Scan(&count)
 	if err == nil {
 		m := sqlutil.NewMigrator(db)
 		m.AddMigrations(sqlutil.Migration{
@@ -72,6 +75,16 @@ func NewPostgresKeyChangesTable(db *sql.DB) (tables.KeyChanges, error) {
 			Up:      deltas.UpRefactorKeyChanges,
 		})
 		return s, m.Up(context.Background())
+	} else {
+		switch e := err.(type) {
+		case *pq.Error:
+			// ignore undefined_column (42703) errors, as this is expected at this point
+			if e.Code != "42703" {
+				return nil, err
+			}
+		default:
+			return nil, err
+		}
 	}
 	return s, nil
 }

keyserver/storage/sqlite3/key_changes_table.go

@@ -61,7 +61,8 @@ func NewSqliteKeyChangesTable(db *sql.DB) (tables.KeyChanges, error) {
 	// TODO: Remove when we are sure we are not having goose artefacts in the db
 	// This forces an error, which indicates the migration is already applied, since the
 	// column partition was removed from the table
-	err = db.QueryRow("SELECT partition FROM keyserver_key_changes LIMIT 1;").Scan()
+	var count int
+	err = db.QueryRow("SELECT partition FROM keyserver_key_changes LIMIT 1;").Scan(&count)
 	if err == nil {
 		m := sqlutil.NewMigrator(db)
 		m.AddMigrations(sqlutil.Migration{

roomserver/storage/postgres/storage.go

@@ -19,8 +19,10 @@ import (
 	"database/sql"
 	"fmt"
 
+	"github.com/lib/pq"
 	// Import the postgres database driver.
 	_ "github.com/lib/pq"
+
 	"github.com/matrix-org/dendrite/internal/caching"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/roomserver/storage/postgres/deltas"
@@ -53,21 +55,32 @@ func Open(base *base.BaseDendrite, dbProperties *config.DatabaseOptions, cache c
 	// TODO: Remove when we are sure we are not having goose artefacts in the db
 	// This forces an error, which indicates the migration is already applied, since the
 	// column event_nid was removed from the table
-	err = db.QueryRow("SELECT event_nid FROM roomserver_state_block LIMIT 1;").Scan()
+	var eventNID int
+	err = db.QueryRow("SELECT event_nid FROM roomserver_state_block LIMIT 1;").Scan(&eventNID)
 	if err == nil {
 		m := sqlutil.NewMigrator(db)
 		m.AddMigrations(sqlutil.Migration{
 			Version: "roomserver: state blocks refactor",
 			Up:      deltas.UpStateBlocksRefactor,
 		})
-		if err := m.Up(base.Context()); err != nil {
+		if err = m.Up(base.Context()); err != nil {
+			return nil, err
+		}
+	} else {
+		switch e := err.(type) {
+		case *pq.Error:
+			// ignore undefined_column (42703) errors, as this is expected at this point
+			if e.Code != "42703" {
+				return nil, err
+			}
+		default:
 			return nil, err
 		}
 	}
 
 	// Then prepare the statements. Now that the migrations have run, any columns referred
 	// to in the database code should now exist.
-	if err := d.prepare(db, writer, cache); err != nil {
+	if err = d.prepare(db, writer, cache); err != nil {
 		return nil, err
 	}
 

roomserver/storage/sqlite3/storage.go

@@ -20,6 +20,8 @@ import (
 	"database/sql"
 	"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"
@@ -27,7 +29,6 @@ import (
 	"github.com/matrix-org/dendrite/roomserver/types"
 	"github.com/matrix-org/dendrite/setup/base"
 	"github.com/matrix-org/dendrite/setup/config"
-	"github.com/matrix-org/gomatrixserverlib"
 )
 
 // A Database is used to store room events and stream offsets.
@@ -63,21 +64,22 @@ func Open(base *base.BaseDendrite, dbProperties *config.DatabaseOptions, cache c
 	// TODO: Remove when we are sure we are not having goose artefacts in the db
 	// This forces an error, which indicates the migration is already applied, since the
 	// column event_nid was removed from the table
-	err = db.QueryRow("SELECT event_nid FROM roomserver_state_block LIMIT 1;").Scan()
+	var eventNID int
+	err = db.QueryRow("SELECT event_nid FROM roomserver_state_block LIMIT 1;").Scan(&eventNID)
 	if err == nil {
 		m := sqlutil.NewMigrator(db)
 		m.AddMigrations(sqlutil.Migration{
 			Version: "roomserver: state blocks refactor",
 			Up:      deltas.UpStateBlocksRefactor,
 		})
-		if err := m.Up(base.Context()); err != nil {
+		if err = m.Up(base.Context()); err != nil {
 			return nil, err
 		}
 	}
 
 	// Then prepare the statements. Now that the migrations have run, any columns referred
 	// to in the database code should now exist.
-	if err := d.prepare(db, writer, cache); err != nil {
+	if err = d.prepare(db, writer, cache); err != nil {
 		return nil, err
 	}
 

syncapi/consumers/clientapi.go

@@ -21,6 +21,11 @@ import (
 	"fmt"
 
 	"github.com/getsentry/sentry-go"
+	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/nats-io/nats.go"
+	"github.com/sirupsen/logrus"
+	log "github.com/sirupsen/logrus"
+
 	"github.com/matrix-org/dendrite/internal/eventutil"
 	"github.com/matrix-org/dendrite/setup/config"
 	"github.com/matrix-org/dendrite/setup/jetstream"
@@ -29,10 +34,6 @@ import (
 	"github.com/matrix-org/dendrite/syncapi/producers"
 	"github.com/matrix-org/dendrite/syncapi/storage"
 	"github.com/matrix-org/dendrite/syncapi/types"
-	"github.com/matrix-org/gomatrixserverlib"
-	"github.com/nats-io/nats.go"
-	"github.com/sirupsen/logrus"
-	log "github.com/sirupsen/logrus"
 )
 
 // OutputClientDataConsumer consumes events that originated in the client API server.
@@ -107,7 +108,8 @@ func (s *OutputClientDataConsumer) onMessage(ctx context.Context, msg *nats.Msg)
 			"type":       output.Type,
 			"room_id":    output.RoomID,
 			log.ErrorKey: err,
-		}).Panicf("could not save account data")
+		}).Errorf("could not save account data")
+		return false
 	}
 
 	if err = s.sendReadUpdate(ctx, userID, output); err != nil {