Skip to content

Commit e8d453a

Browse files
raymondjacobsonraymondjacobson
committed
perf: reduce notification read replica pressure
1 parent 39da256 commit e8d453a

3 files changed

Lines changed: 260 additions & 28 deletions

File tree

‎api/v1_notifications.go‎

Lines changed: 166 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,12 @@
11
package api
22

33
import (
4+
"context"
45
"encoding/json"
6+
"errors"
57
"slices"
68
"strings"
9+
"time"
710

811
"api.audius.co/api/dbv1"
912
"api.audius.co/trashid"
@@ -21,14 +24,32 @@ import (
2124
// surfaces every target — only the actor list is bounded by this cap.
2225
const notificationRelatedActorsPerGroup = 1
2326

27+
const (
28+
// Keep notification reads well below the production replica's
29+
// max_standby_streaming_delay so pathological reads fail fast instead of
30+
// holding hot-standby snapshots and making the replica fall behind.
31+
notificationReadTimeout = 8 * time.Second
32+
33+
// Historically `limit=0` fell back to the default limit of 20, so unread
34+
// polling counted at most the first page of notification groups.
35+
notificationUnreadPollLimit = 20
36+
)
37+
2438
type GetNotificationsQueryParams struct {
25-
// Note that when limit is 0, we return 20 items to calculate unread count
2639
Limit int `query:"limit" default:"20" validate:"min=0,max=100"`
2740
Types []string `query:"types" validate:"dive,oneof=announcement follow repost save remix cosign create tip_receive tip_send challenge_reward repost_of_repost save_of_repost tastemaker reaction supporter_dethroned supporter_rank_up supporting_rank_up milestone track_added_to_playlist tier_change trending trending_playlist trending_underground usdc_purchase_buyer usdc_purchase_seller track_added_to_purchased_album request_manager approve_manager_request track_collaborator_invite track_collaborator_accept claimable_reward comment comment_thread comment_mention comment_reaction listen_streak_reminder fan_remix_contest_started fan_remix_contest_ended fan_remix_contest_ending_soon fan_remix_contest_winners_selected fan_remix_contest_submission artist_remix_contest_ended artist_remix_contest_ending_soon artist_remix_contest_submissions fan_club_text_post remix_contest_update"`
2841
GroupID string `query:"group_id" validate:"omitempty"`
2942
Timestamp float64 `query:"timestamp" validate:"omitempty,min=0"`
3043
}
3144

45+
type notificationRow struct {
46+
Type string `json:"type"`
47+
GroupID string `json:"group_id"`
48+
Actions []json.RawMessage `json:"actions"`
49+
IsSeen bool `json:"is_seen"`
50+
SeenAt interface{} `json:"seen_at"`
51+
}
52+
3253
var unsupportedNotificationTypes = []string{
3354
// No frontend support
3455
"usdc_transfer",
@@ -43,12 +64,87 @@ var unsupportedNotificationTypes = []string{
4364
"remix",
4465
}
4566

67+
func (app *ApiServer) v1NotificationsUnreadPoll(c *fiber.Ctx, params GetNotificationsQueryParams) error {
68+
sql := `
69+
WITH latest_seen AS (
70+
SELECT MAX(seen_at) AS seen_at
71+
FROM notification_seen
72+
WHERE user_id = @user_id
73+
),
74+
-- Equivalent to ARRAY[@user_id] && n.user_ids, split so single-recipient rows
75+
-- can use notification_single_recipient_user_timestamp_idx while
76+
-- multi-recipient arrays keep the existing overlap semantics.
77+
matched_notifications AS (
78+
SELECT n.type, n.group_id
79+
FROM notification n
80+
CROSS JOIN latest_seen
81+
WHERE array_length(n.user_ids, 1) = 1
82+
AND n.user_ids[1] = @user_id
83+
AND n.timestamp > (now()::timestamp - interval '90 days')
84+
AND n.timestamp > COALESCE(latest_seen.seen_at, '-infinity'::timestamp)
85+
AND (n.type = ANY(@types) OR @types IS NULL)
86+
AND (n.type != ALL(@unsupported_types))
87+
UNION ALL
88+
SELECT n.type, n.group_id
89+
FROM notification n
90+
CROSS JOIN latest_seen
91+
WHERE COALESCE(array_length(n.user_ids, 1), 0) != 1
92+
AND ARRAY[@user_id] && n.user_ids
93+
AND n.timestamp > (now()::timestamp - interval '90 days')
94+
AND n.timestamp > COALESCE(latest_seen.seen_at, '-infinity'::timestamp)
95+
AND (n.type = ANY(@types) OR @types IS NULL)
96+
AND (n.type != ALL(@unsupported_types))
97+
)
98+
SELECT COUNT(*)
99+
FROM (
100+
SELECT 1
101+
FROM matched_notifications
102+
GROUP BY type, group_id
103+
LIMIT @limit
104+
) unread_notifications;
105+
`
106+
ctx, cancel := context.WithTimeout(c.Context(), notificationReadTimeout)
107+
defer cancel()
108+
109+
var unreadCount int
110+
err := app.pool.QueryRow(ctx, sql, pgx.NamedArgs{
111+
"user_id": app.getUserId(c),
112+
"limit": notificationUnreadPollLimit,
113+
"types": params.Types,
114+
"unsupported_types": unsupportedNotificationTypes,
115+
}).Scan(&unreadCount)
116+
if err != nil {
117+
if errors.Is(err, context.DeadlineExceeded) {
118+
return fiber.NewError(fiber.StatusGatewayTimeout, "notifications unread query timed out")
119+
}
120+
return err
121+
}
122+
123+
return c.JSON(fiber.Map{
124+
"data": fiber.Map{
125+
"notifications": []notificationRow{},
126+
"unread_count": unreadCount,
127+
},
128+
"related": fiber.Map{
129+
"users": []any{},
130+
"tracks": []any{},
131+
"playlists": []any{},
132+
},
133+
})
134+
}
135+
46136
func (app *ApiServer) v1Notifications(c *fiber.Ctx) error {
137+
limitZeroPoll := c.Query("limit") == "0"
138+
47139
params := GetNotificationsQueryParams{}
48140
if err := app.ParseAndValidateQueryParams(c, &params); err != nil {
49141
return err
50142
}
51143

144+
if limitZeroPoll {
145+
return app.v1NotificationsUnreadPoll(c, params)
146+
}
147+
52148
sql := `
53149
-- user_seen is a window function that gets windows between seen events.
54150
--
@@ -68,6 +164,62 @@ WITH user_seen as (
68164
ORDER BY
69165
seen_at desc
70166
LIMIT 10
167+
),
168+
-- Equivalent to ARRAY[@user_id] && n.user_ids, split so single-recipient rows
169+
-- can use notification_single_recipient_user_timestamp_idx while
170+
-- multi-recipient arrays keep the existing overlap semantics.
171+
matched_notifications AS (
172+
SELECT
173+
n.id,
174+
n.specifier,
175+
n.group_id,
176+
n.type,
177+
n.timestamp,
178+
n.data
179+
FROM notification n
180+
WHERE
181+
array_length(n.user_ids, 1) = 1
182+
AND n.user_ids[1] = @user_id
183+
AND (n.type = ANY(@types) OR @types IS NULL)
184+
AND (n.type != ALL(@unsupported_types))
185+
AND (
186+
-- Initial load: bound to the last 90 days so heavy users don't fan out
187+
-- over their entire notification history. Pagination (timestamp_offset > 0)
188+
-- is unbounded so scrolling further back still works.
189+
(@timestamp_offset = 0 AND @group_id_offset = '' AND n.timestamp > (now()::timestamp - interval '90 days')) OR
190+
(@timestamp_offset = 0 AND @group_id_offset != '' AND n.group_id < @group_id_offset) OR
191+
(@timestamp_offset > 0 AND n.timestamp < to_timestamp(@timestamp_offset)) OR
192+
(
193+
@group_id_offset != '' AND @timestamp_offset > 0 AND
194+
(n.timestamp = to_timestamp(@timestamp_offset) AND n.group_id < @group_id_offset)
195+
)
196+
)
197+
UNION ALL
198+
SELECT
199+
n.id,
200+
n.specifier,
201+
n.group_id,
202+
n.type,
203+
n.timestamp,
204+
n.data
205+
FROM notification n
206+
WHERE
207+
COALESCE(array_length(n.user_ids, 1), 0) != 1
208+
AND ARRAY[@user_id] && n.user_ids
209+
AND (n.type = ANY(@types) OR @types IS NULL)
210+
AND (n.type != ALL(@unsupported_types))
211+
AND (
212+
-- Initial load: bound to the last 90 days so heavy users don't fan out
213+
-- over their entire notification history. Pagination (timestamp_offset > 0)
214+
-- is unbounded so scrolling further back still works.
215+
(@timestamp_offset = 0 AND @group_id_offset = '' AND n.timestamp > (now()::timestamp - interval '90 days')) OR
216+
(@timestamp_offset = 0 AND @group_id_offset != '' AND n.group_id < @group_id_offset) OR
217+
(@timestamp_offset > 0 AND n.timestamp < to_timestamp(@timestamp_offset)) OR
218+
(
219+
@group_id_offset != '' AND @timestamp_offset > 0 AND
220+
(n.timestamp = to_timestamp(@timestamp_offset) AND n.group_id < @group_id_offset)
221+
)
222+
)
71223
)
72224
SELECT
73225
n.type,
@@ -103,7 +255,7 @@ SELECT
103255
ELSE null
104256
END AS seen_at
105257
FROM
106-
notification n
258+
matched_notifications n
107259
LEFT JOIN user_seen ON
108260
user_seen.seen_at >= n.timestamp AND user_seen.prev_seen_at < n.timestamp
109261
-- Join with tracks table to filter out deleted tracks for "create" notifications that have track_id
@@ -125,12 +277,8 @@ LEFT JOIN track_collaborators tc ON
125277
tc.track_id = (n.data->>'track_id')::integer AND
126278
tc.collaborator_user_id = (n.data->>'collaborator_user_id')::integer
127279
WHERE
128-
(ARRAY[@user_id] && n.user_ids)
129-
AND (n.type = ANY(@types) OR @types IS NULL)
130-
-- Ignore notification types not supported by frontend
131-
AND (n.type != ALL(@unsupported_types))
132280
-- Filter out notifications for deleted tracks (only for create notifications that have track_id)
133-
AND (
281+
(
134282
n.type != 'create'
135283
OR NOT (n.data ? 'track_id')
136284
OR (t.is_delete = false AND t.is_unlisted = false)
@@ -204,18 +352,6 @@ WHERE
204352
)
205353
)
206354
)
207-
AND (
208-
-- Initial load: bound to the last 90 days so heavy users don't fan out
209-
-- over their entire notification history. Pagination (timestamp_offset > 0)
210-
-- is unbounded so scrolling further back still works.
211-
(@timestamp_offset = 0 AND @group_id_offset = '' AND n.timestamp > (now()::timestamp - interval '90 days')) OR
212-
(@timestamp_offset = 0 AND @group_id_offset != '' AND n.group_id < @group_id_offset) OR
213-
(@timestamp_offset > 0 AND n.timestamp < to_timestamp(@timestamp_offset)) OR
214-
(
215-
@group_id_offset != '' AND @timestamp_offset > 0 AND
216-
(n.timestamp = to_timestamp(@timestamp_offset) AND n.group_id < @group_id_offset)
217-
)
218-
)
219355
GROUP BY
220356
n.type, n.group_id, user_seen.seen_at, user_seen.prev_seen_at,
221357
CASE
@@ -234,15 +370,11 @@ limit @limit::int
234370
;
235371
`
236372
userId := app.getUserId(c)
237-
type GetNotifsRow struct {
238-
Type string `json:"type"`
239-
GroupID string `json:"group_id"`
240-
Actions []json.RawMessage `json:"actions"`
241-
IsSeen bool `json:"is_seen"`
242-
SeenAt interface{} `json:"seen_at"`
243-
}
244373

245-
rows, err := app.pool.Query(c.Context(), sql, pgx.NamedArgs{
374+
ctx, cancel := context.WithTimeout(c.Context(), notificationReadTimeout)
375+
defer cancel()
376+
377+
rows, err := app.pool.Query(ctx, sql, pgx.NamedArgs{
246378
"user_id": userId,
247379
"limit": params.Limit,
248380
"types": params.Types,
@@ -251,11 +383,17 @@ limit @limit::int
251383
"unsupported_types": unsupportedNotificationTypes,
252384
})
253385
if err != nil {
386+
if errors.Is(err, context.DeadlineExceeded) {
387+
return fiber.NewError(fiber.StatusGatewayTimeout, "notifications query timed out")
388+
}
254389
return err
255390
}
256391

257-
notifs, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByNameLax[GetNotifsRow])
392+
notifs, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByNameLax[notificationRow])
258393
if err != nil {
394+
if errors.Is(err, context.DeadlineExceeded) {
395+
return fiber.NewError(fiber.StatusGatewayTimeout, "notifications query timed out")
396+
}
259397
return err
260398
}
261399

‎api/v1_notifications_test.go‎

Lines changed: 78 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -74,6 +74,84 @@ func TestV1Notifications(t *testing.T) {
7474
})
7575
}
7676

77+
func TestV1Notifications_LimitZeroReturnsUnreadOnly(t *testing.T) {
78+
app := emptyTestApp(t)
79+
80+
fixtures := database.FixtureMap{
81+
"notification": []map[string]any{
82+
{
83+
"id": 1,
84+
"specifier": "111",
85+
"group_id": "tip_send:user_id:111:signature:eee",
86+
"type": "tip_send",
87+
"user_ids": []int{1},
88+
"data": []byte(`{"amount": 100000000, "tx_signature": "asdf", "sender_user_id": 111, "receiver_user_id": 222}`),
89+
},
90+
},
91+
}
92+
93+
database.Seed(app.pool.Replicas[0], fixtures)
94+
95+
status, body := testGet(t, app, "/v1/notifications/"+trashid.MustEncodeHashID(1)+"?limit=0")
96+
assert.Equal(t, 200, status)
97+
98+
jsonAssert(t, body, map[string]any{
99+
"data.notifications.#": 0,
100+
"data.unread_count": 1,
101+
"related.users.#": 0,
102+
"related.tracks.#": 0,
103+
"related.playlists.#": 0,
104+
})
105+
}
106+
107+
func TestV1Notifications_ReturnsMultiRecipientRows(t *testing.T) {
108+
app := emptyTestApp(t)
109+
110+
fixtures := database.FixtureMap{
111+
"notification": []map[string]any{
112+
{
113+
"id": 1,
114+
"specifier": "single",
115+
"group_id": "single:1",
116+
"type": "milestone",
117+
"user_ids": []int{1},
118+
"timestamp": time.Now().Add(-1 * time.Minute),
119+
"data": []byte(`{"type": "TRACK_REPOST_COUNT", "threshold": 10, "track_id": 101}`),
120+
},
121+
{
122+
"id": 2,
123+
"specifier": "multi",
124+
"group_id": "multi:1",
125+
"type": "milestone",
126+
"user_ids": []int{1, 2},
127+
"timestamp": time.Now(),
128+
"data": []byte(`{"type": "TRACK_SAVE_COUNT", "threshold": 10, "track_id": 102}`),
129+
},
130+
},
131+
}
132+
133+
database.Seed(app.pool.Replicas[0], fixtures)
134+
135+
status, body := testGet(t, app, "/v1/notifications/"+trashid.MustEncodeHashID(1))
136+
assert.Equal(t, 200, status)
137+
138+
jsonAssert(t, body, map[string]any{
139+
"data.notifications.#": 2,
140+
"data.notifications.0.type": "milestone",
141+
"data.notifications.0.group_id": "multi:1",
142+
"data.notifications.1.type": "milestone",
143+
"data.notifications.1.group_id": "single:1",
144+
})
145+
146+
status, body = testGet(t, app, "/v1/notifications/"+trashid.MustEncodeHashID(1)+"?limit=0")
147+
assert.Equal(t, 200, status)
148+
149+
jsonAssert(t, body, map[string]any{
150+
"data.notifications.#": 0,
151+
"data.unread_count": 2,
152+
})
153+
}
154+
77155
func TestV1Notifications_NotDeletedTrack(t *testing.T) {
78156
app := emptyTestApp(t)
79157

‎ddl/migrations/0227_notification_single_recipient_user_timestamp_idx.sql‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
-- Production stats show notification rows are overwhelmingly single-recipient:
2+
-- user_ids usually contains exactly one recipient. Querying those common rows
3+
-- through the broad GIN array index is expensive for high-fanout users, and the
4+
-- planner can fall back to full table scans. Give the API a btree path keyed by
5+
-- recipient and recency.
6+
--
7+
-- NOTE: intentionally NOT wrapped in BEGIN/COMMIT so CREATE INDEX
8+
-- CONCURRENTLY can run without holding an ACCESS EXCLUSIVE lock on notification.
9+
-- IF NOT EXISTS makes the migration idempotent.
10+
11+
CREATE INDEX CONCURRENTLY IF NOT EXISTS notification_single_recipient_user_timestamp_idx
12+
ON public.notification ((user_ids[1]), "timestamp" DESC, group_id DESC, type)
13+
WHERE array_length(user_ids, 1) = 1;
14+
15+
COMMENT ON INDEX public.notification_single_recipient_user_timestamp_idx IS
16+
'Covers notification reads for the common single-recipient user_ids array path.';

0 commit comments

Comments
 (0)