Skip to content

Bound initial notification reads #989

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
raymondjacobson merged 1 commit into from
Jul 7, 2026
Merged

Bound initial notification reads #989

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
196 changes: 161 additions & 35 deletions api/v1_notifications.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,9 +34,15 @@ const (
// polling counted at most the first page of notification groups.
notificationUnreadPollLimit = 20

// Timestamp pagination should only need the newest rows before the cursor.
// This keeps hot users from grouping years of notifications before LIMIT.
notificationPaginationCandidateLimit = 25000
// Notification reads should only need the newest rows in the requested
// window. This keeps hot users from grouping years of notifications before
// LIMIT.
notificationReadCandidateLimit = 25000

// Keep the expensive validation/JSON aggregation work bounded after the
// newest notification groups have been selected.
notificationGroupCandidateLimit = 200
notificationActionsPerGroupLimit = 100
)

type GetNotificationsQueryParams struct {
Expand Down Expand Up @@ -78,8 +84,8 @@ WITH latest_seen AS (
-- Equivalent to ARRAY[@user_id] && n.user_ids, split so single-recipient rows
-- can use notification_single_recipient_user_timestamp_idx while
-- multi-recipient arrays keep the existing overlap semantics.
single_recipient_groups AS MATERIALIZED (
SELECT n.type, n.group_id
single_recipient_candidates AS MATERIALIZED (
SELECT n.type, n.group_id, n.timestamp
FROM notification n
CROSS JOIN latest_seen
WHERE array_length(n.user_ids, 1) = 1
Expand All @@ -88,7 +94,13 @@ single_recipient_groups AS MATERIALIZED (
AND n.timestamp > COALESCE(latest_seen.seen_at, '-infinity'::timestamp)
AND (n.type = ANY(@types) OR @types IS NULL)
AND (n.type != ALL(@unsupported_types))
GROUP BY n.type, n.group_id
ORDER BY n.timestamp DESC, n.group_id DESC
LIMIT @candidate_limit
),
single_recipient_groups AS MATERIALIZED (
SELECT type, group_id
FROM single_recipient_candidates
GROUP BY type, group_id
LIMIT @limit
),
single_recipient_count AS MATERIALIZED (
Expand All @@ -97,8 +109,8 @@ single_recipient_count AS MATERIALIZED (
),
-- Only touch the broader multi-recipient GIN path if the fast single-recipient
-- path did not already satisfy the capped unread count.
multi_recipient_groups AS MATERIALIZED (
SELECT n.type, n.group_id
multi_recipient_candidates AS MATERIALIZED (
SELECT n.type, n.group_id, n.timestamp
FROM notification n
CROSS JOIN latest_seen
WHERE (SELECT unread_count FROM single_recipient_count) < @limit
Expand All @@ -114,7 +126,13 @@ multi_recipient_groups AS MATERIALIZED (
WHERE sg.type = n.type
AND sg.group_id = n.group_id
)
GROUP BY n.type, n.group_id
ORDER BY n.timestamp DESC, n.group_id DESC
LIMIT @candidate_limit
),
multi_recipient_groups AS MATERIALIZED (
SELECT type, group_id
FROM multi_recipient_candidates
GROUP BY type, group_id
LIMIT GREATEST(@limit - (SELECT unread_count FROM single_recipient_count), 0)
)
SELECT COUNT(*)
Expand All @@ -129,12 +147,15 @@ FROM (
defer cancel()

var unreadCount int
err := app.pool.QueryRow(ctx, sql, pgx.NamedArgs{
"user_id": app.getUserId(c),
"limit": notificationUnreadPollLimit,
"types": params.Types,
"unsupported_types": unsupportedNotificationTypes,
}).Scan(&unreadCount)
err := app.v1NotificationsReadTx(ctx, func(tx pgx.Tx) error {
return tx.QueryRow(ctx, sql, pgx.NamedArgs{
"user_id": app.getUserId(c),
"limit": notificationUnreadPollLimit,
"candidate_limit": notificationReadCandidateLimit,
"types": params.Types,
"unsupported_types": unsupportedNotificationTypes,
}).Scan(&unreadCount)
})
if err != nil {
if errors.Is(err, context.DeadlineExceeded) {
return fiber.NewError(fiber.StatusGatewayTimeout, "notifications unread query timed out")
Expand Down Expand Up @@ -167,6 +188,7 @@ func (app *ApiServer) v1Notifications(c *fiber.Ctx) error {
return app.v1NotificationsUnreadPoll(c, params)
}

usesCandidateLimit := params.Timestamp > 0 || (params.Timestamp == 0 && params.GroupID == "")
matchedNotificationsCTE := `
-- Equivalent to ARRAY[@user_id] && n.user_ids, split so single-recipient rows
-- can use notification_single_recipient_user_timestamp_idx while
Expand Down Expand Up @@ -226,7 +248,75 @@ matched_notifications AS (
)
`

if params.Timestamp > 0 {
if params.Timestamp == 0 && params.GroupID == "" {
matchedNotificationsCTE = `
-- Initial loads are bounded to 90 days, but hot users can still have enough
-- rows in that window to time out while grouping. Pull newest candidates from
-- each recipient shape first, then apply the same grouping/filtering.
single_recipient_candidates AS MATERIALIZED (
SELECT
n.id,
n.specifier,
n.group_id,
n.type,
n.timestamp,
n.data
FROM notification n
WHERE
array_length(n.user_ids, 1) = 1
AND n.user_ids[1] = @user_id
AND (n.type = ANY(@types) OR @types IS NULL)
AND (n.type != ALL(@unsupported_types))
AND n.timestamp > (now()::timestamp - interval '90 days')
ORDER BY n.timestamp DESC, n.group_id DESC
LIMIT @candidate_limit
),
multi_recipient_candidates AS MATERIALIZED (
SELECT
n.id,
n.specifier,
n.group_id,
n.type,
n.timestamp,
n.data
FROM notification n
WHERE
COALESCE(array_length(n.user_ids, 1), 0) != 1
AND ARRAY[@user_id] && n.user_ids
AND (n.type = ANY(@types) OR @types IS NULL)
AND (n.type != ALL(@unsupported_types))
AND n.timestamp > (now()::timestamp - interval '90 days')
ORDER BY n.timestamp DESC, n.group_id DESC
LIMIT @candidate_limit
),
matched_notifications AS (
SELECT * FROM single_recipient_candidates
UNION ALL
SELECT * FROM multi_recipient_candidates
),
candidate_groups AS MATERIALIZED (
SELECT type, group_id, max(timestamp) AS max_timestamp
FROM matched_notifications
GROUP BY type, group_id
ORDER BY max(timestamp) DESC, group_id DESC
LIMIT @group_candidate_limit
),
capped_notifications AS MATERIALIZED (
SELECT id, specifier, group_id, type, timestamp, data
FROM (
SELECT
n.*,
row_number() OVER (
PARTITION BY n.type, n.group_id
ORDER BY n.timestamp DESC, n.id DESC
) AS action_rank
FROM matched_notifications n
JOIN candidate_groups g USING (type, group_id)
) ranked
WHERE action_rank <= @actions_per_group_limit
)
`
} else if params.Timestamp > 0 {
matchedNotificationsCTE = `
-- Timestamp pagination can otherwise aggregate every older notification for
-- prolific users before LIMIT. Pull a generous timestamp-ordered candidate set
Expand Down Expand Up @@ -254,7 +344,7 @@ single_recipient_candidates AS MATERIALIZED (
)
)
ORDER BY n.timestamp DESC, n.group_id DESC
LIMIT @pagination_candidate_limit
LIMIT @candidate_limit
),
multi_recipient_candidates AS MATERIALIZED (
SELECT
Expand All @@ -279,16 +369,42 @@ multi_recipient_candidates AS MATERIALIZED (
)
)
ORDER BY n.timestamp DESC, n.group_id DESC
LIMIT @pagination_candidate_limit
LIMIT @candidate_limit
),
matched_notifications AS (
SELECT * FROM single_recipient_candidates
UNION ALL
SELECT * FROM multi_recipient_candidates
),
candidate_groups AS MATERIALIZED (
SELECT type, group_id, max(timestamp) AS max_timestamp
FROM matched_notifications
GROUP BY type, group_id
ORDER BY max(timestamp) DESC, group_id DESC
LIMIT @group_candidate_limit
),
capped_notifications AS MATERIALIZED (
SELECT id, specifier, group_id, type, timestamp, data
FROM (
SELECT
n.*,
row_number() OVER (
PARTITION BY n.type, n.group_id
ORDER BY n.timestamp DESC, n.id DESC
) AS action_rank
FROM matched_notifications n
JOIN candidate_groups g USING (type, group_id)
) ranked
WHERE action_rank <= @actions_per_group_limit
)
`
}

notificationSourceCTE := "matched_notifications"
if usesCandidateLimit {
notificationSourceCTE = "capped_notifications"
}

sql := `
-- user_seen is a window function that gets windows between seen events.
--
Expand Down Expand Up @@ -344,7 +460,7 @@ SELECT
ELSE null
END AS seen_at
FROM
matched_notifications n
` + notificationSourceCTE + ` n
LEFT JOIN user_seen ON
user_seen.seen_at >= n.timestamp AND user_seen.prev_seen_at < n.timestamp
-- Join with tracks table to filter out deleted tracks for "create" notifications that have track_id
Expand Down Expand Up @@ -475,14 +591,16 @@ limit @limit::int
"unsupported_types": unsupportedNotificationTypes,
}

if params.Timestamp > 0 {
args["pagination_candidate_limit"] = notificationPaginationCandidateLimit
if usesCandidateLimit {
args["candidate_limit"] = notificationReadCandidateLimit
args["group_candidate_limit"] = notificationGroupCandidateLimit
args["actions_per_group_limit"] = notificationActionsPerGroupLimit
}

var notifs []*notificationRow
var err error
if params.Timestamp > 0 {
notifs, err = app.v1NotificationsPaginatedRows(ctx, sql, args)
if usesCandidateLimit {
notifs, err = app.v1NotificationsRowsWithReadTx(ctx, sql, args)
} else {
rows, queryErr := app.pool.Query(ctx, sql, args)
if queryErr != nil {
Expand Down Expand Up @@ -605,15 +723,15 @@ limit @limit::int

}

func (app *ApiServer) v1NotificationsPaginatedRows(ctx context.Context, sql string, args pgx.NamedArgs) ([]*notificationRow, error) {
func (app *ApiServer) v1NotificationsReadTx(ctx context.Context, fn func(pgx.Tx) error) error {
pool := dbv1.ChooseReplica(app.pool.Replicas)
if pool == nil {
return nil, pgx.ErrNoRows
return pgx.ErrNoRows
}

tx, err := pool.BeginTx(ctx, pgx.TxOptions{AccessMode: pgx.ReadOnly})
if err != nil {
return nil, err
return err
}
defer func() {
rollbackCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
Expand All @@ -624,20 +742,28 @@ func (app *ApiServer) v1NotificationsPaginatedRows(ctx context.Context, sql stri
// The production planner overestimates the multi-recipient overlap branch
// and prefers a table scan; keep the override scoped to this read only.
if _, err := tx.Exec(ctx, "SET LOCAL enable_seqscan = off"); err != nil {
return nil, err
return err
}

rows, err := tx.Query(ctx, sql, args)
if err != nil {
return nil, err
if err := fn(tx); err != nil {
return err
}

notifs, err := pgx.CollectRows(rows, pgx.RowToAddrOfStructByNameLax[notificationRow])
if err != nil {
return nil, err
}
return tx.Commit(ctx)
}

if err := tx.Commit(ctx); err != nil {
func (app *ApiServer) v1NotificationsRowsWithReadTx(ctx context.Context, sql string, args pgx.NamedArgs) ([]*notificationRow, error) {
var notifs []*notificationRow
err := app.v1NotificationsReadTx(ctx, func(tx pgx.Tx) error {
rows, err := tx.Query(ctx, sql, args)
if err != nil {
return err
}

notifs, err = pgx.CollectRows(rows, pgx.RowToAddrOfStructByNameLax[notificationRow])
return err
})
if err != nil {
return nil, err
}

Expand Down
Loading