From 7a10bedea53b4dd07ab211e1b08b153e23ddbb9b Mon Sep 17 00:00:00 2001 From: tajniak81 <13187254+tajniak81@users.noreply.github.com> Date: Tue, 3 Nov 2026 09:46:35 +0100 Subject: [PATCH] PocketBase store: query rows are closed by defer, as lint asks Three reads closed their rows by hand, early, because the call writes right after or loops over chunks; any other early return would have leaked them, and golangci-lint's sqlclosecheck failed CI on it. Each read is now a function of its own with a deferred Close, as sqlstore already does, so the rows are still closed before the writes. Co-Authored-By: Claude Opus 5.5 --- server/internal/storage/pocketbase/keys.go | 81 +++++++++++++-------- server/internal/storage/pocketbase/rooms.go | 43 +++++++---- 2 files changed, 78 insertions(+), 46 deletions(-) diff --git a/server/internal/storage/pocketbase/keys.go b/server/internal/storage/pocketbase/keys.go index b37a3b9..877e194 100644 --- a/server/internal/storage/pocketbase/keys.go +++ b/server/internal/storage/pocketbase/keys.go @@ -99,27 +99,38 @@ func (s *Keys) DeviceListChanges(ctx context.Context, userIDs []string, from, to args = append(args, to) } - rows, err := s.db.QueryContext(ctx, query, args...) + changes, err := s.deviceListChanges(ctx, query, args) if err != nil { - return nil, fmt.Errorf("read device list changes: %w", err) - } - for rows.Next() { - var c storage.DeviceListChange - if err := rows.Scan(&c.UserID, &c.StreamPos); err != nil { - _ = rows.Close() - return nil, fmt.Errorf("read device list changes: %w", err) - } - out = append(out, c) - } - _ = rows.Close() - if err := rows.Err(); err != nil { - return nil, fmt.Errorf("read device list changes: %w", err) + return nil, err } + out = append(out, changes...) } slices.SortFunc(out, func(a, b storage.DeviceListChange) int { return cmp.Compare(a.StreamPos, b.StreamPos) }) return out, nil } +// deviceListChanges runs one chunk of DeviceListChanges. +func (s *Keys) deviceListChanges(ctx context.Context, query string, args []any) ([]storage.DeviceListChange, error) { + rows, err := s.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, fmt.Errorf("read device list changes: %w", err) + } + defer func() { _ = rows.Close() }() + + var out []storage.DeviceListChange + for rows.Next() { + var c storage.DeviceListChange + if err := rows.Scan(&c.UserID, &c.StreamPos); err != nil { + return nil, fmt.Errorf("read device list changes: %w", err) + } + out = append(out, c) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("read device list changes: %w", err) + } + return out, nil +} + // DeviceListPosition implements storage.KeyStore. func (s *Keys) DeviceListPosition(ctx context.Context, userID string) (int64, error) { var pos int64 @@ -257,23 +268,9 @@ func (s *Keys) AddOneTimeKeys(ctx context.Context, userID, deviceID string, keys for i, k := range keys { ids[i] = k.KeyID } - held := map[string]string{} - rows, err := t.query(`SELECT key_id, key_json FROM keyserver_one_time_keys - WHERE user_id = ? AND device_id = ? AND key_id IN `+inList, userID, deviceID, jsonList(ids)) + held, err := heldOneTimeKeys(t, userID, deviceID, ids) if err != nil { - return fmt.Errorf("read the one-time keys of %s/%s: %w", userID, deviceID, err) - } - for rows.Next() { - var id, keyJSON string - if err := rows.Scan(&id, &keyJSON); err != nil { - _ = rows.Close() - return fmt.Errorf("read the one-time keys of %s/%s: %w", userID, deviceID, err) - } - held[id] = keyJSON - } - _ = rows.Close() - if err := rows.Err(); err != nil { - return fmt.Errorf("read the one-time keys of %s/%s: %w", userID, deviceID, err) + return err } for _, k := range keys { @@ -293,6 +290,30 @@ func (s *Keys) AddOneTimeKeys(ctx context.Context, userID, deviceID string, keys }) } +// heldOneTimeKeys reads which of ids the device already holds, by key ID. +// The rows are closed when it returns, before the call writes. +func heldOneTimeKeys(t *tx, userID, deviceID string, ids []string) (map[string]string, error) { + rows, err := t.query(`SELECT key_id, key_json FROM keyserver_one_time_keys + WHERE user_id = ? AND device_id = ? AND key_id IN `+inList, userID, deviceID, jsonList(ids)) + if err != nil { + return nil, fmt.Errorf("read the one-time keys of %s/%s: %w", userID, deviceID, err) + } + defer func() { _ = rows.Close() }() + + held := map[string]string{} + for rows.Next() { + var id, keyJSON string + if err := rows.Scan(&id, &keyJSON); err != nil { + return nil, fmt.Errorf("read the one-time keys of %s/%s: %w", userID, deviceID, err) + } + held[id] = keyJSON + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("read the one-time keys of %s/%s: %w", userID, deviceID, err) + } + return held, nil +} + // OneTimeKeyCounts implements storage.KeyStore. func (s *Keys) OneTimeKeyCounts(ctx context.Context, userID, deviceID string) (map[string]int, error) { rows, err := s.db.QueryContext(ctx, `SELECT algorithm, COUNT(*) FROM keyserver_one_time_keys diff --git a/server/internal/storage/pocketbase/rooms.go b/server/internal/storage/pocketbase/rooms.go index 7cc49a0..714000e 100644 --- a/server/internal/storage/pocketbase/rooms.go +++ b/server/internal/storage/pocketbase/rooms.go @@ -501,22 +501,9 @@ func (w *roomWrite) applyStateDelta(rw storage.RoomWrite) error { func (w *roomWrite) recountMembers(rw storage.RoomWrite) error { joined := map[string]bool{} if !rw.StateDeltaIsWholeState { - rows, err := w.query(`SELECT state_key FROM roomserver_current_state - WHERE room_id = ? AND event_type = 'm.room.member' AND membership = 'join'`, rw.Event.RoomID) - if err != nil { - return fmt.Errorf("recount joined members: %w", err) - } - for rows.Next() { - var user string - if err := rows.Scan(&user); err != nil { - _ = rows.Close() - return fmt.Errorf("recount joined members: %w", err) - } - joined[user] = true - } - _ = rows.Close() - if err := rows.Err(); err != nil { - return fmt.Errorf("recount joined members: %w", err) + var err error + if joined, err = w.storedJoins(rw.Event.RoomID); err != nil { + return err } } for _, e := range rw.StateDelta { @@ -537,6 +524,30 @@ func (w *roomWrite) recountMembers(rw storage.RoomWrite) error { return nil } +// storedJoins reads the users the stored current state has joined to the +// room. The rows are closed when it returns, before the call writes. +func (w *roomWrite) storedJoins(roomID string) (map[string]bool, error) { + rows, err := w.query(`SELECT state_key FROM roomserver_current_state + WHERE room_id = ? AND event_type = 'm.room.member' AND membership = 'join'`, roomID) + if err != nil { + return nil, fmt.Errorf("recount joined members: %w", err) + } + defer func() { _ = rows.Close() }() + + joined := map[string]bool{} + for rows.Next() { + var user string + if err := rows.Scan(&user); err != nil { + return nil, fmt.Errorf("recount joined members: %w", err) + } + joined[user] = true + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("recount joined members: %w", err) + } + return joined, nil +} + func (w *roomWrite) replaceExtremities(roomID string, extremities []string) error { if len(extremities) == 0 { return nil