perf(feed): rewrite v_feed, bound feed_delta, server-time SSE cursor (H6,H7,M9)
H6: migration 006 replaces v_feed's double LEFT JOIN + COUNT(DISTINCT) (which materialized a likes×comments Cartesian per upload) with correlated scalar subqueries that each use their own index. Output columns are unchanged, so feed + hashtag-filtered paths both benefit. H7: feed_delta now applies the feed rate limit (keyed per user), caps results at 200 rows, and clamps how far back a client `since` may reach (7 days). When clamped or capped it returns reload_required=true; the SSE client turns that into a full feed reload instead of streaming the whole gallery through the view on every tab refocus. M9: the SSE reconnect cursor is now advanced only from server timestamps (an upload's created_at, seeded from the feed and updated on new-upload events and delta responses), never the client clock — so a skewed phone clock can't drop events missed while backgrounded. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
23
backend/migrations/006_feed_view_perf.down.sql
Normal file
23
backend/migrations/006_feed_view_perf.down.sql
Normal file
@@ -0,0 +1,23 @@
|
|||||||
|
-- Restore the original join-based v_feed definition.
|
||||||
|
CREATE OR REPLACE VIEW v_feed AS
|
||||||
|
SELECT
|
||||||
|
u.id,
|
||||||
|
u.event_id,
|
||||||
|
u.user_id,
|
||||||
|
usr.display_name AS uploader_name,
|
||||||
|
usr.is_banned,
|
||||||
|
usr.uploads_hidden,
|
||||||
|
u.preview_path,
|
||||||
|
u.thumbnail_path,
|
||||||
|
u.mime_type,
|
||||||
|
u.caption,
|
||||||
|
u.created_at,
|
||||||
|
COUNT(DISTINCT l.user_id) AS like_count,
|
||||||
|
COUNT(DISTINCT c.id) AS comment_count
|
||||||
|
FROM upload u
|
||||||
|
JOIN "user" usr ON u.user_id = usr.id
|
||||||
|
LEFT JOIN "like" l ON l.upload_id = u.id
|
||||||
|
LEFT JOIN comment c ON c.upload_id = u.id AND c.deleted_at IS NULL
|
||||||
|
WHERE u.deleted_at IS NULL
|
||||||
|
AND usr.uploads_hidden = FALSE
|
||||||
|
GROUP BY u.id, usr.display_name, usr.is_banned, usr.uploads_hidden;
|
||||||
26
backend/migrations/006_feed_view_perf.up.sql
Normal file
26
backend/migrations/006_feed_view_perf.up.sql
Normal file
@@ -0,0 +1,26 @@
|
|||||||
|
-- H6: replace v_feed's double LEFT JOIN + COUNT(DISTINCT) (which materializes a
|
||||||
|
-- likes×comments Cartesian per upload before de-duping) with correlated scalar
|
||||||
|
-- subqueries. Each count now uses its own index (idx_like_upload /
|
||||||
|
-- idx_comment_upload) and there is no GROUP BY. The output columns are
|
||||||
|
-- unchanged, so every consumer keeps working.
|
||||||
|
CREATE OR REPLACE VIEW v_feed AS
|
||||||
|
SELECT
|
||||||
|
u.id,
|
||||||
|
u.event_id,
|
||||||
|
u.user_id,
|
||||||
|
usr.display_name AS uploader_name,
|
||||||
|
usr.is_banned,
|
||||||
|
usr.uploads_hidden,
|
||||||
|
u.preview_path,
|
||||||
|
u.thumbnail_path,
|
||||||
|
u.mime_type,
|
||||||
|
u.caption,
|
||||||
|
u.created_at,
|
||||||
|
(SELECT COUNT(*) FROM "like" l
|
||||||
|
WHERE l.upload_id = u.id) AS like_count,
|
||||||
|
(SELECT COUNT(*) FROM comment c
|
||||||
|
WHERE c.upload_id = u.id AND c.deleted_at IS NULL) AS comment_count
|
||||||
|
FROM upload u
|
||||||
|
JOIN "user" usr ON u.user_id = usr.id
|
||||||
|
WHERE u.deleted_at IS NULL
|
||||||
|
AND usr.uploads_hidden = FALSE;
|
||||||
@@ -190,31 +190,70 @@ pub struct DeltaQuery {
|
|||||||
pub struct DeltaResponse {
|
pub struct DeltaResponse {
|
||||||
pub uploads: Vec<FeedUpload>,
|
pub uploads: Vec<FeedUpload>,
|
||||||
pub deleted_ids: Vec<Uuid>,
|
pub deleted_ids: Vec<Uuid>,
|
||||||
|
/// Set when the delta was clamped (too-old cursor) or hit the row cap — the
|
||||||
|
/// client should do a full feed reload instead of trusting the partial set.
|
||||||
|
pub reload_required: bool,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Hard cap on how many uploads one delta returns. Beyond this the client is
|
||||||
|
/// told to reload rather than streaming the whole gallery through the view.
|
||||||
|
const DELTA_LIMIT: i64 = 200;
|
||||||
|
/// How far back a client-supplied `since` may reach. A tab backgrounded for days
|
||||||
|
/// must not pull the entire event on reconnect.
|
||||||
|
const DELTA_MAX_LOOKBACK_DAYS: i64 = 7;
|
||||||
|
|
||||||
pub async fn feed_delta(
|
pub async fn feed_delta(
|
||||||
State(state): State<AppState>,
|
State(state): State<AppState>,
|
||||||
auth: AuthUser,
|
auth: AuthUser,
|
||||||
Query(q): Query<DeltaQuery>,
|
Query(q): Query<DeltaQuery>,
|
||||||
) -> Result<Json<DeltaResponse>, AppError> {
|
) -> Result<Json<DeltaResponse>, AppError> {
|
||||||
|
// H7: feed_delta runs the (expensive) feed query and fires on every tab
|
||||||
|
// refocus, so it needs the same rate limit as feed(), keyed per user.
|
||||||
|
let rate_limits_on = config::get_bool(&state.pool, "rate_limits_enabled", true).await;
|
||||||
|
let feed_rate_on = config::get_bool(&state.pool, "feed_rate_enabled", true).await;
|
||||||
|
if rate_limits_on && feed_rate_on {
|
||||||
|
let rate_limit = config::get_usize(&state.pool, "feed_rate_per_min", 60).await;
|
||||||
|
if !state.rate_limiter.check(
|
||||||
|
format!("feed_delta:{}", auth.user_id),
|
||||||
|
rate_limit,
|
||||||
|
Duration::from_secs(60),
|
||||||
|
) {
|
||||||
|
return Err(AppError::TooManyRequests(
|
||||||
|
"Zu viele Anfragen. Bitte warte kurz und versuche es erneut.".into(),
|
||||||
|
None,
|
||||||
|
));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Clamp the lookback server-side; signal a full reload if we had to.
|
||||||
|
let min_since = Utc::now() - chrono::Duration::days(DELTA_MAX_LOOKBACK_DAYS);
|
||||||
|
let clamped = q.since < min_since;
|
||||||
|
let since = if clamped { min_since } else { q.since };
|
||||||
|
|
||||||
let rows = sqlx::query_as::<_, FeedRow>(
|
let rows = sqlx::query_as::<_, FeedRow>(
|
||||||
"SELECT id, user_id, uploader_name, preview_path, thumbnail_path,
|
"SELECT id, user_id, uploader_name, preview_path, thumbnail_path,
|
||||||
mime_type, caption, like_count, comment_count, created_at
|
mime_type, caption, like_count, comment_count, created_at
|
||||||
FROM v_feed
|
FROM v_feed
|
||||||
WHERE event_id = $1 AND created_at > $2
|
WHERE event_id = $1 AND created_at > $2
|
||||||
ORDER BY created_at DESC",
|
ORDER BY created_at DESC
|
||||||
|
LIMIT $3",
|
||||||
)
|
)
|
||||||
.bind(auth.event_id)
|
.bind(auth.event_id)
|
||||||
.bind(q.since)
|
.bind(since)
|
||||||
|
.bind(DELTA_LIMIT + 1)
|
||||||
.fetch_all(&state.pool)
|
.fetch_all(&state.pool)
|
||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
|
let capped = rows.len() as i64 > DELTA_LIMIT;
|
||||||
|
let rows: Vec<FeedRow> = rows.into_iter().take(DELTA_LIMIT as usize).collect();
|
||||||
|
let reload_required = clamped || capped;
|
||||||
|
|
||||||
let deleted_ids: Vec<(Uuid,)> = sqlx::query_as(
|
let deleted_ids: Vec<(Uuid,)> = sqlx::query_as(
|
||||||
"SELECT id FROM upload
|
"SELECT id FROM upload
|
||||||
WHERE event_id = $1 AND deleted_at IS NOT NULL AND deleted_at > $2",
|
WHERE event_id = $1 AND deleted_at IS NOT NULL AND deleted_at > $2",
|
||||||
)
|
)
|
||||||
.bind(auth.event_id)
|
.bind(auth.event_id)
|
||||||
.bind(q.since)
|
.bind(since)
|
||||||
.fetch_all(&state.pool)
|
.fetch_all(&state.pool)
|
||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
@@ -234,6 +273,7 @@ pub async fn feed_delta(
|
|||||||
Ok(Json(DeltaResponse {
|
Ok(Json(DeltaResponse {
|
||||||
uploads,
|
uploads,
|
||||||
deleted_ids: deleted_ids.into_iter().map(|r| r.0).collect(),
|
deleted_ids: deleted_ids.into_iter().map(|r| r.0).collect(),
|
||||||
|
reload_required,
|
||||||
}))
|
}))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -50,7 +50,18 @@ const KNOWN_EVENTS = [
|
|||||||
* Synthetic event types — not emitted by the server, dispatched locally to fan out
|
* Synthetic event types — not emitted by the server, dispatched locally to fan out
|
||||||
* cross-cutting state changes (e.g. delta-fetch results after a reconnect).
|
* cross-cutting state changes (e.g. delta-fetch results after a reconnect).
|
||||||
*/
|
*/
|
||||||
export type SyntheticEvent = 'feed-delta';
|
export type SyntheticEvent = 'feed-delta' | 'feed-reload';
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Advance the reconnect cursor using a **server** timestamp (an upload's
|
||||||
|
* `created_at`), never the client clock. A phone clock skewed ahead would
|
||||||
|
* otherwise make the cursor jump past events that happened while the tab was
|
||||||
|
* backgrounded. ISO-8601 UTC strings compare correctly lexicographically.
|
||||||
|
*/
|
||||||
|
function noteServerTime(ts: string | null | undefined): void {
|
||||||
|
if (!ts) return;
|
||||||
|
if (!lastEventTime || ts > lastEventTime) lastEventTime = ts;
|
||||||
|
}
|
||||||
|
|
||||||
export function onSseEvent(eventType: string, handler: EventHandler): () => void {
|
export function onSseEvent(eventType: string, handler: EventHandler): () => void {
|
||||||
if (!handlers.has(eventType)) {
|
if (!handlers.has(eventType)) {
|
||||||
@@ -93,12 +104,13 @@ export function connectSse(): void {
|
|||||||
eventSource.onopen = () => {
|
eventSource.onopen = () => {
|
||||||
// Successful connection — reset the backoff counter.
|
// Successful connection — reset the backoff counter.
|
||||||
reconnectAttempt = 0;
|
reconnectAttempt = 0;
|
||||||
// If we have a previous timestamp this is a reconnect — fetch the gap.
|
// If we have a previous (server-derived) timestamp this is a reconnect
|
||||||
|
// — fetch the gap. The cursor is only ever advanced from server
|
||||||
|
// timestamps (noteServerTime), so client clock skew can't drop events.
|
||||||
const since = lastEventTime;
|
const since = lastEventTime;
|
||||||
if (since) {
|
if (since) {
|
||||||
void deltaFetchAndFan(since);
|
void deltaFetchAndFan(since);
|
||||||
}
|
}
|
||||||
lastEventTime = new Date().toISOString();
|
|
||||||
};
|
};
|
||||||
|
|
||||||
for (const eventName of KNOWN_EVENTS) {
|
for (const eventName of KNOWN_EVENTS) {
|
||||||
@@ -146,7 +158,15 @@ export function setLastEventTime(time: string): void {
|
|||||||
}
|
}
|
||||||
|
|
||||||
function dispatch(eventType: string, data: string): void {
|
function dispatch(eventType: string, data: string): void {
|
||||||
lastEventTime = new Date().toISOString();
|
// Advance the cursor from the server timestamp carried by a new upload.
|
||||||
|
// Other event types don't carry one and must not bump it off the client clock.
|
||||||
|
if (eventType === 'new-upload') {
|
||||||
|
try {
|
||||||
|
noteServerTime((JSON.parse(data) as { created_at?: string }).created_at);
|
||||||
|
} catch {
|
||||||
|
// payload not JSON — ignore
|
||||||
|
}
|
||||||
|
}
|
||||||
const list = handlers.get(eventType);
|
const list = handlers.get(eventType);
|
||||||
if (list) {
|
if (list) {
|
||||||
for (const handler of list) {
|
for (const handler of list) {
|
||||||
@@ -166,6 +186,14 @@ async function deltaFetchAndFan(since: string): Promise<void> {
|
|||||||
const response = await api.get<DeltaResponse>(
|
const response = await api.get<DeltaResponse>(
|
||||||
`/feed/delta?since=${encodeURIComponent(since)}`
|
`/feed/delta?since=${encodeURIComponent(since)}`
|
||||||
);
|
);
|
||||||
|
// Advance the cursor from the newest server timestamp in the delta.
|
||||||
|
for (const u of response.uploads) noteServerTime(u.created_at);
|
||||||
|
// The server clamped the window or hit the row cap — the partial delta
|
||||||
|
// can't be trusted, so ask the page to do a full reload instead.
|
||||||
|
if (response.reload_required) {
|
||||||
|
dispatch('feed-reload', '{}');
|
||||||
|
return;
|
||||||
|
}
|
||||||
dispatch('feed-delta', JSON.stringify(response));
|
dispatch('feed-delta', JSON.stringify(response));
|
||||||
} catch {
|
} catch {
|
||||||
// non-fatal
|
// non-fatal
|
||||||
|
|||||||
@@ -28,6 +28,7 @@ export interface FeedResponse {
|
|||||||
export interface DeltaResponse {
|
export interface DeltaResponse {
|
||||||
uploads: FeedUpload[];
|
uploads: FeedUpload[];
|
||||||
deleted_ids: string[];
|
deleted_ids: string[];
|
||||||
|
reload_required: boolean;
|
||||||
}
|
}
|
||||||
|
|
||||||
// mirrors backend/src/handlers/feed.rs::HashtagCount
|
// mirrors backend/src/handlers/feed.rs::HashtagCount
|
||||||
|
|||||||
@@ -2,7 +2,7 @@
|
|||||||
import { goto } from '$app/navigation';
|
import { goto } from '$app/navigation';
|
||||||
import { getToken, getUserId } from '$lib/auth';
|
import { getToken, getUserId } from '$lib/auth';
|
||||||
import { api } from '$lib/api';
|
import { api } from '$lib/api';
|
||||||
import { connectSse, disconnectSse, onSseEvent } from '$lib/sse';
|
import { connectSse, disconnectSse, onSseEvent, setLastEventTime } from '$lib/sse';
|
||||||
import { onMount, onDestroy } from 'svelte';
|
import { onMount, onDestroy } from 'svelte';
|
||||||
import FeedGrid from '$lib/components/FeedGrid.svelte';
|
import FeedGrid from '$lib/components/FeedGrid.svelte';
|
||||||
import FeedListCard from '$lib/components/FeedListCard.svelte';
|
import FeedListCard from '$lib/components/FeedListCard.svelte';
|
||||||
@@ -200,6 +200,9 @@
|
|||||||
}),
|
}),
|
||||||
onSseEvent('like-update', () => loadFeed(true)),
|
onSseEvent('like-update', () => loadFeed(true)),
|
||||||
onSseEvent('new-comment', () => loadFeed(true)),
|
onSseEvent('new-comment', () => loadFeed(true)),
|
||||||
|
// Delta was clamped/capped server-side (H7) — do a full reload instead
|
||||||
|
// of trusting a partial set.
|
||||||
|
onSseEvent('feed-reload', () => loadFeed(true)),
|
||||||
// Synthetic event from the SSE client after a foreground reconnect — merge
|
// Synthetic event from the SSE client after a foreground reconnect — merge
|
||||||
// any uploads + deletions we missed while the tab was hidden.
|
// any uploads + deletions we missed while the tab was hidden.
|
||||||
onSseEvent('feed-delta', (data) => {
|
onSseEvent('feed-delta', (data) => {
|
||||||
@@ -243,6 +246,10 @@
|
|||||||
const res = await api.get<FeedResponse>(`/feed?${params}`);
|
const res = await api.get<FeedResponse>(`/feed?${params}`);
|
||||||
uploads = res.uploads;
|
uploads = res.uploads;
|
||||||
nextCursor = res.next_cursor;
|
nextCursor = res.next_cursor;
|
||||||
|
// Seed the SSE reconnect cursor from the newest server timestamp so the
|
||||||
|
// delta on the next reconnect is based on server time, not the client
|
||||||
|
// clock (M9).
|
||||||
|
if (uploads.length) setLastEventTime(uploads[0].created_at);
|
||||||
} catch (e) {
|
} catch (e) {
|
||||||
// Initial / user-triggered refresh is worth surfacing — background SSE refetches are noisier and silenced below.
|
// Initial / user-triggered refresh is worth surfacing — background SSE refetches are noisier and silenced below.
|
||||||
if (!refresh) toastError(e);
|
if (!refresh) toastError(e);
|
||||||
|
|||||||
Reference in New Issue
Block a user