More `rows.Close()` and `rows.Err()` (#3262)

Looks like we missed some `rows.Close()`

Even though `rows.Err()` is mostly not necessary, we should be more
consistent in the DB layer.

[skip ci]

11 files changed, 18 insertions, 12 deletions

diff --git a/syncapi/storage/postgres/current_room_state_table.go b/syncapi/storage/postgres/current_room_state_table.go
index b0148bef..ec0b27ad 100644
--- a/syncapi/storage/postgres/current_room_state_table.go
+++ b/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) {
diff --git a/syncapi/storage/postgres/memberships_table.go b/syncapi/storage/postgres/memberships_table.go
index 4fe4260d..e5208b89 100644
--- a/syncapi/storage/postgres/memberships_table.go
+++ b/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
 	)
diff --git a/syncapi/storage/postgres/peeks_table.go b/syncapi/storage/postgres/peeks_table.go
index 64183073..1120dce0 100644
--- a/syncapi/storage/postgres/peeks_table.go
+++ b/syncapi/storage/postgres/peeks_table.go
@@ -164,7 +164,7 @@ func (s *peekStatements) SelectPeekingDevices(
 		devices = append(devices, types.PeekingDevice{UserID: userID, DeviceID: deviceID})
 		result[roomID] = devices
 	}
-	return result, nil
+	return result, rows.Err()
 }
 
 func (s *peekStatements) SelectMaxPeekID(
diff --git a/syncapi/storage/postgres/presence_table.go b/syncapi/storage/postgres/presence_table.go
index f37b5331..53acecce 100644
--- a/syncapi/storage/postgres/presence_table.go
+++ b/syncapi/storage/postgres/presence_table.go
@@ -144,7 +144,7 @@ func (p *presenceStatements) GetPresenceForUsers(
 		presence.ClientFields.Presence = presence.Presence.String()
 		result = append(result, presence)
 	}
-	return result, err
+	return result, rows.Err()
 }
 
 func (p *presenceStatements) GetMaxPresenceID(ctx context.Context, txn *sql.Tx) (pos types.StreamPosition, err error) {
diff --git a/syncapi/storage/sqlite3/current_room_state_table.go b/syncapi/storage/sqlite3/current_room_state_table.go
index 78b2e397..f430fcc0 100644
--- a/syncapi/storage/sqlite3/current_room_state_table.go
+++ b/syncapi/storage/sqlite3/current_room_state_table.go
@@ -177,7 +177,7 @@ func (s *currentRoomStateStatements) SelectJoinedUsers(
 		users = append(users, userID)
 		result[roomID] = users
 	}
-	return result, nil
+	return result, rows.Err()
 }
 
 // SelectJoinedUsersInRoom returns a map of room ID to a list of joined user IDs for a given room.
@@ -236,7 +236,7 @@ func (s *currentRoomStateStatements) SelectRoomIDsWithMembership(
 		}
 		result = append(result, roomID)
 	}
-	return result, nil
+	return result, rows.Err()
 }
 
 // SelectRoomIDsWithAnyMembership returns a map of all memberships for the given user.
@@ -419,7 +419,7 @@ func currentRoomStateRowsToStreamEvents(rows *sql.Rows) ([]types.StreamEvent, er
 		})
 	}
 
-	return events, nil
+	return events, rows.Err()
 }
 
 func rowsToEvents(rows *sql.Rows) ([]*rstypes.HeaderedEvent, error) {
diff --git a/syncapi/storage/sqlite3/invites_table.go b/syncapi/storage/sqlite3/invites_table.go
index ebb469d2..e50b5cbf 100644
--- a/syncapi/storage/sqlite3/invites_table.go
+++ b/syncapi/storage/sqlite3/invites_table.go
@@ -176,7 +176,7 @@ func (s *inviteEventsStatements) SelectInviteEventsInRange(
 	if lastPos == 0 {
 		lastPos = r.To
 	}
-	return result, retired, lastPos, nil
+	return result, retired, lastPos, rows.Err()
 }
 
 func (s *inviteEventsStatements) SelectMaxInviteID(
diff --git a/syncapi/storage/sqlite3/memberships_table.go b/syncapi/storage/sqlite3/memberships_table.go
index a1b16306..9e50422e 100644
--- a/syncapi/storage/sqlite3/memberships_table.go
+++ b/syncapi/storage/sqlite3/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"
@@ -163,6 +164,7 @@ func (s *membershipsStatements) SelectMemberships(
 	if err != nil {
 		return
 	}
+	defer internal.CloseAndLogIfError(ctx, rows, "SelectMemberships: failed to close rows")
 	var eventID string
 	for rows.Next() {
 		if err = rows.Scan(&eventID); err != nil {
diff --git a/syncapi/storage/sqlite3/output_room_events_table.go b/syncapi/storage/sqlite3/output_room_events_table.go
index 93caee80..c7b11d3e 100644
--- a/syncapi/storage/sqlite3/output_room_events_table.go
+++ b/syncapi/storage/sqlite3/output_room_events_table.go
@@ -274,7 +274,7 @@ func (s *outputRoomEventsStatements) SelectStateInRange(
 		}
 	}
 
-	return stateNeeded, eventIDToEvent, nil
+	return stateNeeded, eventIDToEvent, rows.Err()
 }
 
 // MaxID returns the ID of the last inserted event in this table. 'txn' is optional. If it is not supplied,
@@ -520,7 +520,7 @@ func rowsToStreamEvents(rows *sql.Rows) ([]types.StreamEvent, error) {
 			ExcludeFromSync: excludeFromSync,
 		})
 	}
-	return result, nil
+	return result, rows.Err()
 }
 func (s *outputRoomEventsStatements) SelectContextEvent(
 	ctx context.Context, txn *sql.Tx, roomID, eventID string,
diff --git a/syncapi/storage/sqlite3/output_room_events_topology_table.go b/syncapi/storage/sqlite3/output_room_events_topology_table.go
index 36967d1e..c00fb7a7 100644
--- a/syncapi/storage/sqlite3/output_room_events_topology_table.go
+++ b/syncapi/storage/sqlite3/output_room_events_topology_table.go
@@ -18,6 +18,7 @@ import (
 	"context"
 	"database/sql"
 
+	"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"
@@ -137,6 +138,7 @@ func (s *outputRoomEventsTopologyStatements) SelectEventIDsInRange(
 	} else if err != nil {
 		return
 	}
+	defer internal.CloseAndLogIfError(ctx, rows, "SelectEventIDsInRange: failed to close rows")
 
 	// Return the IDs.
 	var eventID string
@@ -155,7 +157,7 @@ func (s *outputRoomEventsTopologyStatements) SelectEventIDsInRange(
 		start = tokens[0]
 		end = tokens[len(tokens)-1]
 	}
-
+	err = rows.Err()
 	return
 }
 
diff --git a/syncapi/storage/sqlite3/peeks_table.go b/syncapi/storage/sqlite3/peeks_table.go
index 5d5200ab..d8998e2b 100644
--- a/syncapi/storage/sqlite3/peeks_table.go
+++ b/syncapi/storage/sqlite3/peeks_table.go
@@ -184,7 +184,7 @@ func (s *peekStatements) SelectPeekingDevices(
 		devices = append(devices, types.PeekingDevice{UserID: userID, DeviceID: deviceID})
 		result[roomID] = devices
 	}
-	return result, nil
+	return result, rows.Err()
 }
 
 func (s *peekStatements) SelectMaxPeekID(
diff --git a/syncapi/storage/sqlite3/presence_table.go b/syncapi/storage/sqlite3/presence_table.go
index 573fbad6..40b57e75 100644
--- a/syncapi/storage/sqlite3/presence_table.go
+++ b/syncapi/storage/sqlite3/presence_table.go
@@ -169,7 +169,7 @@ func (p *presenceStatements) GetPresenceForUsers(
 		presence.ClientFields.Presence = presence.Presence.String()
 		result = append(result, presence)
 	}
-	return result, err
+	return result, rows.Err()
 }
 
 func (p *presenceStatements) GetMaxPresenceID(ctx context.Context, txn *sql.Tx) (pos types.StreamPosition, err error) {