More rows.Close() and rows.Err()

federationapi/storage/postgres/notary_server_keys_metadata_table.go

@@ -151,7 +151,7 @@ func (s *notaryServerKeysMetadataStatements) SelectKeys(ctx context.Context, txn
 		}
 		results = append(results, sk)
 	}
-	return results, nil
+	return results, rows.Err()
 }
 
 func (s *notaryServerKeysMetadataStatements) DeleteOldJSONResponses(ctx context.Context, txn *sql.Tx) error {

federationapi/storage/postgres/queue_json_table.go

@@ -109,5 +109,5 @@ func (s *queueJSONStatements) SelectQueueJSON(
 		}
 		blobs[nid] = blob
 	}
-	return blobs, err
+	return blobs, rows.Err()
 }

federationapi/storage/postgres/relay_servers_table.go

@@ -110,7 +110,7 @@ func (s *relayServersStatements) SelectRelayServers(
 		}
 		result = append(result, spec.ServerName(relayServer))
 	}
-	return result, nil
+	return result, rows.Err()
 }
 
 func (s *relayServersStatements) DeleteRelayServers(

federationapi/storage/sqlite3/joined_hosts_table.go

@@ -216,5 +216,5 @@ func joinedHostsFromStmt(
 		})
 	}
 
-	return result, nil
+	return result, rows.Err()
 }

federationapi/storage/sqlite3/notary_server_keys_metadata_table.go

@@ -154,7 +154,7 @@ func (s *notaryServerKeysMetadataStatements) SelectKeys(ctx context.Context, txn
 		}
 		results = append(results, sk)
 	}
-	return results, nil
+	return results, rows.Err()
 }
 
 func (s *notaryServerKeysMetadataStatements) DeleteOldJSONResponses(ctx context.Context, txn *sql.Tx) error {

federationapi/storage/sqlite3/queue_json_table.go

@@ -135,5 +135,5 @@ func (s *queueJSONStatements) SelectQueueJSON(
 		}
 		blobs[nid] = blob
 	}
-	return blobs, err
+	return blobs, rows.Err()
 }

federationapi/storage/sqlite3/relay_servers_table.go

@@ -109,7 +109,7 @@ func (s *relayServersStatements) SelectRelayServers(
 		}
 		result = append(result, spec.ServerName(relayServer))
 	}
-	return result, nil
+	return result, rows.Err()
 }
 
 func (s *relayServersStatements) DeleteRelayServers(

relayapi/storage/postgres/relay_queue_json_table.go

@@ -109,5 +109,5 @@ func (s *relayQueueJSONStatements) SelectQueueJSON(
 		}
 		blobs[nid] = blob
 	}
-	return blobs, err
+	return blobs, rows.Err()
 }

relayapi/storage/sqlite3/relay_queue_json_table.go

@@ -133,5 +133,5 @@ func (s *relayQueueJSONStatements) SelectQueueJSON(
 		}
 		blobs[nid] = blob
 	}
-	return blobs, err
+	return blobs, rows.Err()
 }

roomserver/storage/postgres/events_table.go

@@ -249,6 +249,7 @@ func (s *eventStatements) BulkSelectSnapshotsFromEventIDs(
 	if err != nil {
 		return nil, err
 	}
+	defer internal.CloseAndLogIfError(ctx, rows, "BulkSelectSnapshotsFromEventIDs: rows.close() failed")
 
 	var eventID string
 	var stateNID types.StateSnapshotNID
@@ -563,7 +564,7 @@ func (s *eventStatements) SelectRoomNIDsForEventNIDs(
 		}
 		result[eventNID] = roomNID
 	}
-	return result, nil
+	return result, rows.Err()
 }
 
 func eventNIDsAsArray(eventNIDs []types.EventNID) pq.Int64Array {

roomserver/storage/postgres/membership_table.go

@@ -363,7 +363,7 @@ func (s *membershipStatements) SelectRoomsWithMembership(
 		}
 		roomNIDs = append(roomNIDs, roomNID)
 	}
-	return roomNIDs, nil
+	return roomNIDs, rows.Err()
 }
 
 func (s *membershipStatements) SelectJoinedUsersSetForRooms(

roomserver/storage/postgres/rooms_table.go

@@ -137,7 +137,7 @@ func (s *roomStatements) SelectRoomIDsWithEvents(ctx context.Context, txn *sql.T
 		}
 		roomIDs = append(roomIDs, roomID)
 	}
-	return roomIDs, nil
+	return roomIDs, rows.Err()
 }
 func (s *roomStatements) InsertRoomNID(
 	ctx context.Context, txn *sql.Tx,
@@ -255,7 +255,7 @@ func (s *roomStatements) SelectRoomVersionsForRoomNIDs(
 		}
 		result[roomNID] = roomVersion
 	}
-	return result, nil
+	return result, rows.Err()
 }
 
 func (s *roomStatements) BulkSelectRoomIDs(ctx context.Context, txn *sql.Tx, roomNIDs []types.RoomNID) ([]string, error) {
@@ -277,7 +277,7 @@ func (s *roomStatements) BulkSelectRoomIDs(ctx context.Context, txn *sql.Tx, roo
 		}
 		roomIDs = append(roomIDs, roomID)
 	}
-	return roomIDs, nil
+	return roomIDs, rows.Err()
 }
 
 func (s *roomStatements) BulkSelectRoomNIDs(ctx context.Context, txn *sql.Tx, roomIDs []string) ([]types.RoomNID, error) {
@@ -299,7 +299,7 @@ func (s *roomStatements) BulkSelectRoomNIDs(ctx context.Context, txn *sql.Tx, ro
 		}
 		roomNIDs = append(roomNIDs, roomNID)
 	}
-	return roomNIDs, nil
+	return roomNIDs, rows.Err()
 }
 
 func roomNIDsAsArray(roomNIDs []types.RoomNID) pq.Int64Array {

roomserver/storage/postgres/user_room_keys_table.go

@@ -162,6 +162,7 @@ func (s *userRoomKeysStatements) SelectAllPublicKeysForUser(ctx context.Context,
 	if errors.Is(err, sql.ErrNoRows) {
 		return nil, nil
 	}
+	defer internal.CloseAndLogIfError(ctx, rows, "SelectAllPublicKeysForUser: failed to close rows")
 
 	resultMap := make(map[types.RoomNID]ed25519.PublicKey)
 
@@ -173,5 +174,5 @@ func (s *userRoomKeysStatements) SelectAllPublicKeysForUser(ctx context.Context,
 		}
 		resultMap[roomNID] = pubkey
 	}
-	return resultMap, err
+	return resultMap, rows.Err()
 }

roomserver/storage/sqlite3/event_json_table.go

@@ -109,5 +109,5 @@ func (s *eventJSONStatements) BulkSelectEventJSON(
 		}
 		result.EventNID = types.EventNID(eventNID)
 	}
-	return results[:i], nil
+	return results[:i], rows.Err()
 }

roomserver/storage/sqlite3/event_state_keys_table.go

@@ -136,7 +136,7 @@ func (s *eventStateKeyStatements) BulkSelectEventStateKeyNID(
 		}
 		result[stateKey] = types.EventStateKeyNID(stateKeyNID)
 	}
-	return result, nil
+	return result, rows.Err()
 }
 
 func (s *eventStateKeyStatements) BulkSelectEventStateKey(
@@ -167,5 +167,5 @@ func (s *eventStateKeyStatements) BulkSelectEventStateKey(
 		}
 		result[types.EventStateKeyNID(stateKeyNID)] = stateKey
 	}
-	return result, nil
+	return result, rows.Err()
 }

roomserver/storage/sqlite3/event_types_table.go

@@ -147,5 +147,5 @@ func (s *eventTypeStatements) BulkSelectEventTypeNID(
 		}
 		result[eventType] = types.EventTypeNID(eventTypeNID)
 	}
-	return result, nil
+	return result, rows.Err()
 }

roomserver/storage/sqlite3/events_table.go

@@ -310,6 +310,9 @@ func (s *eventStatements) BulkSelectStateEventByID(
 		}
 		results = append(results, result)
 	}
+	if err = rows.Err(); err != nil {
+		return nil, err
+	}
 	if !excludeRejected && i != len(eventIDs) {
 		// If there are fewer rows returned than IDs then we were asked to lookup event IDs we don't have.
 		// We don't know which ones were missing because we don't return the string IDs in the query.
@@ -377,7 +380,7 @@ func (s *eventStatements) BulkSelectStateEventByNID(
 			return nil, err
 		}
 	}
-	return results[:i], err
+	return results[:i], rows.Err()
 }
 
 // bulkSelectStateAtEventByID lookups the state at a list of events by event ID.
@@ -425,6 +428,9 @@ func (s *eventStatements) BulkSelectStateAtEventByID(
 			)
 		}
 	}
+	if err = rows.Err(); err != nil {
+		return nil, err
+	}
 	if i != len(eventIDs) {
 		return nil, types.MissingEventError(
 			fmt.Sprintf("storage: event IDs missing from the database (%d != %d)", i, len(eventIDs)),
@@ -507,6 +513,9 @@ func (s *eventStatements) BulkSelectStateAtEventAndReference(
 		result.BeforeStateSnapshotNID = types.StateSnapshotNID(stateSnapshotNID)
 		result.EventID = eventID
 	}
+	if err = rows.Err(); err != nil {
+		return nil, err
+	}
 	if i != len(eventNIDs) {
 		return nil, fmt.Errorf("storage: event NIDs missing from the database (%d != %d)", i, len(eventNIDs))
 	}
@@ -544,6 +553,9 @@ func (s *eventStatements) BulkSelectEventID(ctx context.Context, txn *sql.Tx, ev
 		}
 		results[types.EventNID(eventNID)] = eventID
 	}
+	if err = rows.Err(); err != nil {
+		return nil, err
+	}
 	if i != len(eventNIDs) {
 		return nil, fmt.Errorf("storage: event NIDs missing from the database (%d != %d)", i, len(eventNIDs))
 	}
@@ -602,7 +614,7 @@ func (s *eventStatements) bulkSelectEventNID(ctx context.Context, txn *sql.Tx, e
 			RoomNID:  types.RoomNID(roomNID),
 		}
 	}
-	return results, nil
+	return results, rows.Err()
 }
 
 func (s *eventStatements) SelectMaxEventDepth(ctx context.Context, txn *sql.Tx, eventNIDs []types.EventNID) (int64, error) {
@@ -652,7 +664,7 @@ func (s *eventStatements) SelectRoomNIDsForEventNIDs(
 		}
 		result[eventNID] = roomNID
 	}
-	return result, nil
+	return result, rows.Err()
 }
 
 func eventNIDsAsArray(eventNIDs []types.EventNID) string {

roomserver/storage/sqlite3/invite_table.go

@@ -126,6 +126,9 @@ func (s *inviteStatements) UpdateInviteRetired(
 		}
 		eventIDs = append(eventIDs, inviteEventID)
 	}
+	if err = rows.Err(); err != nil {
+		return
+	}
 	// now retire the invites
 	stmt = sqlutil.TxStmt(txn, s.updateInviteRetiredStmt)
 	_, err = stmt.ExecContext(ctx, roomNID, targetUserNID)
@@ -157,5 +160,5 @@ func (s *inviteStatements) SelectInviteActiveForUserInRoom(
 		result = append(result, types.EventStateKeyNID(senderUserNID))
 		eventIDs = append(eventIDs, eventID)
 	}
-	return result, eventIDs, eventJSON, nil
+	return result, eventIDs, eventJSON, rows.Err()
 }

roomserver/storage/sqlite3/membership_table.go

@@ -250,6 +250,7 @@ func (s *membershipStatements) SelectMembershipsFromRoom(
 		}
 		eventNIDs = append(eventNIDs, eNID)
 	}
+	err = rows.Err()
 	return
 }
 
@@ -277,6 +278,7 @@ func (s *membershipStatements) SelectMembershipsFromRoomAndMembership(
 		}
 		eventNIDs = append(eventNIDs, eNID)
 	}
+	err = rows.Err()
 	return
 }
 
@@ -313,7 +315,7 @@ func (s *membershipStatements) SelectRoomsWithMembership(
 		}
 		roomNIDs = append(roomNIDs, roomNID)
 	}
-	return roomNIDs, nil
+	return roomNIDs, rows.Err()
 }
 
 func (s *membershipStatements) SelectJoinedUsersSetForRooms(ctx context.Context, txn *sql.Tx, roomNIDs []types.RoomNID, userNIDs []types.EventStateKeyNID, localOnly bool) (map[types.EventStateKeyNID]int, error) {

roomserver/storage/sqlite3/room_aliases_table.go

@@ -121,7 +121,7 @@ func (s *roomAliasesStatements) SelectAliasesFromRoomID(
 
 		aliases = append(aliases, alias)
 	}
-
+	err = rows.Err()
 	return
 }
 

roomserver/storage/sqlite3/rooms_table.go

@@ -128,7 +128,7 @@ func (s *roomStatements) SelectRoomIDsWithEvents(ctx context.Context, txn *sql.T
 		}
 		roomIDs = append(roomIDs, roomID)
 	}
-	return roomIDs, nil
+	return roomIDs, rows.Err()
 }
 
 func (s *roomStatements) SelectRoomInfo(ctx context.Context, txn *sql.Tx, roomID string) (*types.RoomInfo, error) {
@@ -265,7 +265,7 @@ func (s *roomStatements) SelectRoomVersionsForRoomNIDs(
 		}
 		result[roomNID] = roomVersion
 	}
-	return result, nil
+	return result, rows.Err()
 }
 
 func (s *roomStatements) BulkSelectRoomIDs(ctx context.Context, txn *sql.Tx, roomNIDs []types.RoomNID) ([]string, error) {
@@ -293,7 +293,7 @@ func (s *roomStatements) BulkSelectRoomIDs(ctx context.Context, txn *sql.Tx, roo
 		}
 		roomIDs = append(roomIDs, roomID)
 	}
-	return roomIDs, nil
+	return roomIDs, rows.Err()
 }
 
 func (s *roomStatements) BulkSelectRoomNIDs(ctx context.Context, txn *sql.Tx, roomIDs []string) ([]types.RoomNID, error) {
@@ -321,5 +321,5 @@ func (s *roomStatements) BulkSelectRoomNIDs(ctx context.Context, txn *sql.Tx, ro
 		}
 		roomNIDs = append(roomNIDs, roomNID)
 	}
-	return roomNIDs, nil
+	return roomNIDs, rows.Err()
 }

roomserver/storage/sqlite3/state_snapshot_table.go

@@ -133,13 +133,16 @@ func (s *stateSnapshotStatements) BulkSelectStateBlockNIDs(
 	var stateBlockNIDsJSON string
 	for ; rows.Next(); i++ {
 		result := &results[i]
-		if err := rows.Scan(&result.StateSnapshotNID, &stateBlockNIDsJSON); err != nil {
+		if err = rows.Scan(&result.StateSnapshotNID, &stateBlockNIDsJSON); err != nil {
 			return nil, err
 		}
-		if err := json.Unmarshal([]byte(stateBlockNIDsJSON), &result.StateBlockNIDs); err != nil {
+		if err = json.Unmarshal([]byte(stateBlockNIDsJSON), &result.StateBlockNIDs); err != nil {
 			return nil, err
 		}
 	}
+	if err = rows.Err(); err != nil {
+		return nil, err
+	}
 	if i != len(stateNIDs) {
 		return nil, types.MissingStateError(fmt.Sprintf("storage: state NIDs missing from the database (%d != %d)", i, len(stateNIDs)))
 	}

roomserver/storage/sqlite3/user_room_keys_table.go

@@ -177,6 +177,7 @@ func (s *userRoomKeysStatements) SelectAllPublicKeysForUser(ctx context.Context,
 	if errors.Is(err, sql.ErrNoRows) {
 		return nil, nil
 	}
+	defer internal.CloseAndLogIfError(ctx, rows, "SelectAllPublicKeysForUser: failed to close rows")
 
 	resultMap := make(map[types.RoomNID]ed25519.PublicKey)
 
@@ -188,5 +189,5 @@ func (s *userRoomKeysStatements) SelectAllPublicKeysForUser(ctx context.Context,
 		}
 		resultMap[roomNID] = pubkey
 	}
-	return resultMap, err
+	return resultMap, rows.Err()
 }

setup/mscs/msc2836/storage.go

@@ -301,7 +301,7 @@ func (p *DB) ChildrenForParent(ctx context.Context, eventID, relType string, rec
 		}
 		children = append(children, evInfo)
 	}
-	return children, nil
+	return children, rows.Err()
 }
 
 func (p *DB) ParentForChild(ctx context.Context, eventID, relType string) (*eventInfo, error) {

syncapi/storage/postgres/current_room_state_table.go

@@ -392,7 +392,7 @@ func currentRoomStateRowsToStreamEvents(rows *sql.Rows) ([]types.StreamEvent, er
 		})
 	}
 
-	return events, nil
+	return events, rows.Err()
 }
 
 func rowsToEvents(rows *sql.Rows) ([]*rstypes.HeaderedEvent, error) {

syncapi/storage/postgres/memberships_table.go

@@ -19,6 +19,7 @@ import (
 	"database/sql"
 	"fmt"
 
+	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
 	rstypes "github.com/matrix-org/dendrite/roomserver/types"
 	"github.com/matrix-org/dendrite/syncapi/storage/tables"
@@ -160,6 +161,7 @@ func (s *membershipsStatements) SelectMemberships(
 	if err != nil {
 		return
 	}
+	defer internal.CloseAndLogIfError(ctx, rows, "SelectMemberships: failed to close rows")
 	var (
 		eventID string
 	)