fabi/EventSnap
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'fix/upload-pipeline-integrity'

Browse Source
This commit is contained in:
fabi
2026-07-28 07:19:25 +02:00
parent d4237ad2ad 05948d8268
commit c4e9b89af0
11 changed files with 365 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
backend/migrations/018_derivatives_rev.down.sql Normal file
View File

@@ -0,0 +1,2 @@
DROP INDEX IF EXISTS idx_upload_derivatives_rev;
ALTER TABLE upload DROP COLUMN IF EXISTS derivatives_rev;

18
backend/migrations/018_derivatives_rev.up.sql Normal file
View File

@@ -0,0 +1,18 @@
-- Track which revision of the derivative pipeline produced an upload's preview/display.
--
-- Rev 1 applies the EXIF orientation tag. Everything generated before it decoded the raw
-- sensor pixels and re-encoded to JPEG (which writes no EXIF), so every portrait phone photo
-- was stored sideways in the feed preview, the diashow display and the keepsake — while the
-- untouched original still rendered upright.
--
-- Existing rows default to 0 so the startup backfill can find and re-generate them exactly
-- once; bump the constant in services/compression.rs if the pipeline ever changes again.
ALTER TABLE upload ADD COLUMN IF NOT EXISTS derivatives_rev SMALLINT NOT NULL DEFAULT 0;
-- Only image derivatives are affected — video thumbnails are extracted by ffmpeg, which
-- already honours the rotation matrix. Mark them current so the backfill skips them.
UPDATE upload SET derivatives_rev = 1 WHERE mime_type NOT LIKE 'image/%';
CREATE INDEX IF NOT EXISTS idx_upload_derivatives_rev
ON upload (derivatives_rev)
WHERE deleted_at IS NULL;

10
backend/src/main.rs
View File

@@ -44,10 +44,12 @@ async fn main() -> Result<()> {
let state = AppState::new(pool.clone(), config.clone());
// Backfill the big-screen display derivative for image uploads processed before it
// existed (v0.17.x). Fire-and-forget behind the compression semaphore; each upload
// falls back to its original in the diashow until its display is generated.
state.compression.backfill_missing_display().await;
// Regenerate image derivatives an older pipeline produced: the big-screen display for
// uploads processed before it existed (v0.17.x), and anything predating the current
// DERIVATIVES_REV (rev 1 applies the EXIF orientation, without which every portrait
// phone photo is stored sideways). Fire-and-forget behind the compression semaphore;
// originals are never touched, so a failure just retries on the next start.
state.compression.backfill_stale_derivatives().await;
// Re-spawn exports for events that were released but whose keepsake never finished
// (crash mid-export). Needs the media/export paths + SSE sender, so it runs here

15
backend/src/models/upload.rs
View File

@@ -148,6 +148,21 @@ impl Upload {
Ok(())
}
/// Stamp which revision of the derivative pipeline produced this row's preview/display,
/// so the startup backfill can find rows generated by an older one exactly once.
pub async fn set_derivatives_rev(
pool: &PgPool,
id: Uuid,
rev: i16,
) -> Result<(), sqlx::Error> {
sqlx::query("UPDATE upload SET derivatives_rev = $2 WHERE id = $1")
.bind(id)
.bind(rev)
.execute(pool)
.await?;
Ok(())
}
pub async fn set_thumbnail_path(
pool: &PgPool,
id: Uuid,

145
backend/src/services/compression.rs
View File

@@ -3,6 +3,7 @@ use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use anyhow::{Context, Result};
use image::ImageDecoder;
use sqlx::PgPool;
use tokio::sync::{Semaphore, broadcast};
use uuid::Uuid;
@@ -48,6 +49,16 @@ impl CompressionWorker {
self.generation.fetch_add(1, Ordering::SeqCst);
}
/// How many times `do_process` is attempted before an upload is given up on. The
/// give-up path is user-visible (the photo disappears), so transient infrastructure
/// errors must not reach it.
const MAX_PROCESS_ATTEMPTS: u32 = 3;
/// Revision of the image-derivative pipeline. Bump this whenever a change makes existing
/// previews/displays wrong, so `backfill_stale_derivatives` regenerates them once on the
/// next start. Rev 1 = EXIF orientation is applied.
const DERIVATIVES_REV: i16 = 1;
/// Spawn a background task to process an uploaded file.
pub fn process(&self, upload_id: Uuid, original_path: String, mime_type: String) {
let worker = self.clone();
@@ -60,10 +71,35 @@ impl CompressionWorker {
if worker.generation.load(Ordering::SeqCst) != born_at {
return;
}
match worker
.do_process(upload_id, &original_path, &mime_type)
.await
{
// Retry before giving up. Most failures here are transient and self-clearing —
// an ENOSPC spike while several guests upload at once, a momentary DB-pool
// exhaustion, a panic inside the image codec — and the give-up path is
// user-visible data loss, so it is worth a few seconds to avoid entering it.
let mut attempt = 1u32;
let outcome = loop {
match worker
.do_process(upload_id, &original_path, &mime_type)
.await
{
Ok(v) => break Ok(v),
Err(e) if attempt < Self::MAX_PROCESS_ATTEMPTS => {
tracing::warn!(
error = ?e, %upload_id, attempt,
"compression attempt failed; retrying"
);
tokio::time::sleep(std::time::Duration::from_secs(2u64.pow(attempt)))
.await;
attempt += 1;
// The data may have been reset while we slept (e2e TRUNCATE).
if worker.generation.load(Ordering::SeqCst) != born_at {
return;
}
}
Err(e) => break Err(e),
}
};
match outcome {
Ok(_) => {
tracing::info!("compression completed for upload {upload_id}");
let _ = worker.sse_tx.send(SseEvent {
@@ -72,21 +108,31 @@ impl CompressionWorker {
});
}
Err(e) => {
tracing::error!("compression failed for upload {upload_id}: {e:#}");
// Auto-cleanup: a failed transcode would otherwise leave a
// permanently broken feed card, silently charge the uploader's
// quota, and orphan the original on disk. Refund + soft-delete
// (one tx, so v_feed excludes it), remove the orphan file, then
// tell the uploader (upload-error toast) and evict the card
// everywhere (upload-deleted, already handled by the feed).
tracing::error!(
"compression failed for upload {upload_id} after {attempt} attempt(s): {e:#}"
);
// Refund + soft-delete (one tx, so v_feed excludes it) so a failed
// transcode doesn't leave a permanently broken feed card or silently
// charge the uploader's quota. Then tell the uploader (upload-error
// toast) and evict the card everywhere (upload-deleted).
//
// The ORIGINAL IS DELIBERATELY KEPT. This path used to `remove_file` it
// unconditionally, which meant any transient error — a disk-full blip
// while saving a derivative, a pool hiccup, a panic in the image codec —
// irreversibly destroyed the guest's only copy of a photo they can never
// retake. The row is only soft-deleted, so keeping the bytes makes the
// upload fully recoverable; the file is orphaned rather than lost, and
// the path is logged so it can be found. `backfill_stale_derivatives`
// already refuses to destroy data on error for exactly this reason.
let _ = Upload::set_compression_status(&worker.pool, upload_id, "failed").await;
if let Err(del) = Upload::soft_delete(&worker.pool, upload_id).await {
tracing::warn!(error = ?del, %upload_id, "failed to soft-delete after compression failure");
}
let orphan = worker.media_path.join(&original_path);
if let Err(rm) = tokio::fs::remove_file(&orphan).await {
tracing::warn!(error = ?rm, path = %orphan.display(), "failed to remove orphaned original");
}
tracing::warn!(
%upload_id,
path = %worker.media_path.join(&original_path).display(),
"original retained for recovery after compression failure"
);
let _ = worker.sse_tx.send(SseEvent {
event_type: "upload-error".to_string(),
data: serde_json::json!({ "upload_id": upload_id, "error": e.to_string() })
@@ -117,6 +163,7 @@ impl CompressionWorker {
.await?;
Upload::set_preview_path(&self.pool, upload_id, &preview_rel).await?;
Upload::set_display_path(&self.pool, upload_id, &display_rel).await?;
Upload::set_derivatives_rev(&self.pool, upload_id, Self::DERIVATIVES_REV).await?;
tracing::info!("preview + display generated for upload {upload_id}");
} else if mime_type.starts_with("video/") {
let thumb_rel = self.generate_video_thumbnail(upload_id, &original).await?;
@@ -172,7 +219,24 @@ impl CompressionWorker {
limits.max_image_height = Some(12_000);
limits.max_alloc = Some(256 * 1024 * 1024);
reader.limits(limits);
let img = reader.decode().context("failed to decode image")?;
// Apply the EXIF orientation. Phones do not rotate the sensor data — they record
// the physical camera orientation in a tag and store the pixels as shot. `decode()`
// hands back those raw pixels, and the JPEG re-encode below writes no EXIF at all,
// so skipping this stores EVERY portrait photo sideways in the feed preview, the
// 2048px diashow display and the keepsake — while the untouched original still
// renders upright, which is why it looks like a viewer bug rather than a pipeline
// one. `into_decoder` carries the limits set above through to the decoder, so the
// decompression-bomb guard is unaffected.
let mut decoder = reader.into_decoder().context("failed to decode image")?;
// A missing or malformed tag is not a failure: most images simply have none.
let orientation = decoder
.orientation()
.unwrap_or(image::metadata::Orientation::NoTransforms);
let mut img =
image::DynamicImage::from_decoder(decoder).context("failed to decode image")?;
img.apply_orientation(orientation);
let img = img;
// Preview: max 800px, preserving aspect ratio (data-saver feed).
img.resize(
@@ -221,33 +285,41 @@ impl CompressionWorker {
))
}
/// One-time backfill: existing image uploads processed before the display derivative
/// existed have a preview but no `display_path`. Regenerate both derivatives for them
/// (decode is cheap and idempotent) and set the path. Unlike the failure path in
/// `process`, a backfill error is logged and skipped — it must NEVER soft-delete an
/// upload that already has a working preview. Fire-and-forget from startup.
pub async fn backfill_missing_display(&self) {
/// Regenerate image derivatives that an older pipeline produced. Fire-and-forget from
/// startup; picks up two cases, both of which leave the ORIGINAL untouched:
///
/// - uploads processed before the `display` derivative existed (preview but no
/// `display_path`), and
/// - uploads whose derivatives predate `DERIVATIVES_REV` — currently rev 1, which applies
/// the EXIF orientation. Everything generated before it is stored sideways for any
/// portrait phone photo.
///
/// Unlike the failure path in `process`, a backfill error is logged and skipped — it must
/// NEVER destroy or soft-delete an upload that already has a working preview.
pub async fn backfill_stale_derivatives(&self) {
let rows = sqlx::query_as::<_, (Uuid, String, String)>(
"SELECT id, original_path, mime_type FROM upload
WHERE display_path IS NULL AND preview_path IS NOT NULL
AND deleted_at IS NULL AND mime_type LIKE 'image/%'",
WHERE deleted_at IS NULL AND mime_type LIKE 'image/%'
AND original_path IS NOT NULL
AND (
(display_path IS NULL AND preview_path IS NOT NULL)
OR derivatives_rev < $1
)",
)
.bind(Self::DERIVATIVES_REV)
.fetch_all(&self.pool)
.await;
let rows = match rows {
Ok(r) => r,
Err(e) => {
tracing::warn!(error = ?e, "display backfill query failed");
tracing::warn!(error = ?e, "derivative backfill query failed");
return;
}
};
if rows.is_empty() {
return;
}
tracing::info!(
"backfilling display derivative for {} upload(s)",
rows.len()
);
tracing::info!("regenerating derivatives for {} upload(s)", rows.len());
for (id, original_path, mime_type) in rows {
let worker = self.clone();
tokio::spawn(async move {
@@ -260,12 +332,19 @@ impl CompressionWorker {
Ok((preview_rel, display_rel)) => {
let _ = Upload::set_preview_path(&worker.pool, id, &preview_rel).await;
let _ = Upload::set_display_path(&worker.pool, id, &display_rel).await;
tracing::info!("display backfilled for upload {id}");
let _ = Upload::set_derivatives_rev(
&worker.pool,
id,
Self::DERIVATIVES_REV,
)
.await;
tracing::info!("derivatives regenerated for upload {id}");
}
Err(e) => {
// Leave the existing preview intact; the diashow falls back to the
// original for this upload until a later successful pass.
tracing::warn!(error = ?e, %id, "display backfill failed; leaving as-is");
// Leave the existing derivatives and the original intact; this row is
// simply retried on the next start. The rev stays behind, which is the
// marker that it still needs doing.
tracing::warn!(error = ?e, %id, "derivative backfill failed; leaving as-is");
}
}
});

21
e2e/fixtures/db.ts
View File

@@ -54,6 +54,27 @@ export const db = {
);
},
async compressionStatus(uploadId: string): Promise<string | null> {
return withClient(async (c) => {
const r = await c.query<{ compression_status: string }>(
`SELECT compression_status FROM upload WHERE id = $1`,
[uploadId]
);
return r.rows[0]?.compression_status ?? null;
});
},
/** Which revision of the derivative pipeline produced this row's preview/display. */
async derivativesRev(uploadId: string): Promise<number | null> {
return withClient(async (c) => {
const r = await c.query<{ derivatives_rev: number }>(
`SELECT derivatives_rev FROM upload WHERE id = $1`,
[uploadId]
);
return r.rows[0]?.derivatives_rev ?? null;
});
},
async countUploadsForUser(userId: string): Promise<number> {
return withClient(async (c) => {
const r = await c.query<{ count: string }>(

80
e2e/specs/02-upload/exif-orientation.spec.ts Normal file
View File

@@ -0,0 +1,80 @@
/**
* Regression guard — EXIF orientation must be applied when generating derivatives.
*
* Phones do not rotate sensor data. They shoot in the sensor's native landscape and record
* how the camera was held in an EXIF `Orientation` tag. `image`'s `decode()` returns the raw
* pixels and ignores that tag, and the JPEG re-encode writes no EXIF at all — so every
* portrait photo was stored SIDEWAYS in the 800px feed preview, the 2048px diashow display
* and the keepsake, while "Original anzeigen" still rendered it upright (the original keeps
* its tag). That asymmetry is why it reads as a viewer bug instead of a pipeline one.
*
* The fixture is 40x20 landscape pixels tagged Orientation=6 ("rotate 90° CW to display"),
* so a correctly-processed derivative is PORTRAIT (20x40). Asserting on the aspect ratio
* rather than the bytes keeps this robust across encoder changes.
*/
import { test, expect } from '../../fixtures/test';
import { uploadRaw } from '../../helpers/upload-client';
import { BASE } from '../../helpers/env';
import { readFileSync } from 'node:fs';
import { join } from 'node:path';
const EXIF_FIXTURE = join(process.cwd(), 'fixtures', 'media', 'portrait-exif6.jpg');
/**
* Read a baseline/progressive JPEG's pixel dimensions from its SOF marker.
* Avoids pulling an image dependency into the suite for one assertion.
*/
function jpegSize(buf: Buffer): { width: number; height: number } {
let i = 2; // skip SOI
while (i < buf.length) {
if (buf[i] !== 0xff) {
i++;
continue;
}
const marker = buf[i + 1];
// SOF0..SOF15, excluding DHT (c4), JPGA (c8) and DAC (cc)
if (marker >= 0xc0 && marker <= 0xcf && marker !== 0xc4 && marker !== 0xc8 && marker !== 0xcc) {
return { height: buf.readUInt16BE(i + 5), width: buf.readUInt16BE(i + 7) };
}
i += 2 + buf.readUInt16BE(i + 2);
}
throw new Error('no SOF marker found — not a JPEG?');
}
test.describe('Upload — EXIF orientation', () => {
test('a rotated photo is upright in the preview and the display derivative', async ({
guest,
db,
}) => {
const g = await guest('SidewaysShooter');
const res = await uploadRaw(g.jwt, readFileSync(EXIF_FIXTURE), {
filename: 'portrait-exif6.jpg',
contentType: 'image/jpeg',
caption: 'hochkant',
});
expect(res.status).toBe(201);
const { id } = (await res.json()) as { id: string };
// Sanity: the SOURCE really is stored landscape with the tag, otherwise this test
// could pass against a pipeline that does nothing.
const source = jpegSize(readFileSync(EXIF_FIXTURE));
expect(source.width).toBeGreaterThan(source.height);
await expect
.poll(() => db.compressionStatus(id), { timeout: 30_000, intervals: [250] })
.toBe('done');
for (const variant of ['preview', 'display'] as const) {
const r = await fetch(`${BASE}/api/v1/upload/${id}/${variant}`, {
headers: { Authorization: `Bearer ${g.jwt}` },
});
expect(r.status, `${variant} must be served`).toBe(200);
const { width, height } = jpegSize(Buffer.from(await r.arrayBuffer()));
expect(
height,
`${variant} must be portrait (${width}x${height}) — EXIF orientation was not applied`
).toBeGreaterThan(width);
}
});
});

84
e2e/specs/02-upload/rejection-visible.spec.ts Normal file
View File

@@ -0,0 +1,84 @@
/**
* Regression guard — a rejected upload must tell the user something.
*
* `UploadQueue.svelte` was 162 lines of complete, working UI — the only renderer of an
* item's error text, the only "Erneut" retry button, the only rate-limit countdown — and it
* was never imported anywhere, so `retryItem`, `removeItem` and `clearCompleted` were all
* unreachable at runtime. On a terminal rejection the store dropped the blob, wrote a clear
* German reason into `entry.error` with the comment "so the UI shows a clear reason", and
* there was no such UI.
*
* Meanwhile the FAB badge counted only pending/uploading, so a rejected photo decremented it
* exactly as if it had succeeded. Net effect: the photo silently vanished — no toast, no
* queue row, no error text, and it never appeared in the feed.
*/
import { test, expect } from '../../fixtures/test';
import { FeedPage, UploadSheet } from '../../page-objects';
import { join } from 'node:path';
const SAMPLE_JPG = join(process.cwd(), 'fixtures', 'media', 'sample.jpg');
test.describe('Upload — a rejected upload is surfaced', () => {
test('a terminally rejected upload toasts, and stays visible in the queue', async ({
page,
api,
host,
guest,
signIn,
}) => {
const g = await guest('RejectedUploader');
await signIn(page, g);
const feed = new FeedPage(page);
const sheet = new UploadSheet(page);
await feed.openUploadSheet();
await sheet.stageFiles([SAMPLE_JPG]);
await sheet.captionInput.waitFor({ state: 'visible', timeout: 10_000 });
// Ban the uploader between staging and sending, so the POST comes back 403 — a
// terminal 4xx the server will keep rejecting, which is the path that purges the blob.
await api.banUser(host.jwt, g.userId);
await sheet.submit();
// 1. The user is told, wherever they are (the flow lands them on /feed).
const toast = page.getByRole('region', { name: 'Benachrichtigungen' });
await expect(toast).toContainText(/sample\.jpg/i, { timeout: 15_000 });
// 2. The queue row survives with its reason and is reachable on /upload — this is what
// the orphaned component made impossible.
await page.goto('/upload');
const queue = page.getByText('Upload-Warteschlange');
await expect(queue, 'the upload queue must be rendered somewhere').toBeVisible({
timeout: 10_000,
});
// Both the status chip ("Gesperrt") and the server's reason ("Du bist gesperrt.") must
// render — the reason is the part that had no UI at all before.
await expect(page.getByText('Gesperrt', { exact: true })).toBeVisible();
await expect(page.getByText('Du bist gesperrt.')).toBeVisible();
// 3. The badge must not read as success. It counted only pending/uploading before, so a
// rejected item dropped it to 0 — indistinguishable from a completed upload.
await expect
.poll(
() =>
page.evaluate(async () => {
return new Promise<number>((resolve, reject) => {
const req = indexedDB.open('eventsnap-uploads', 3);
req.onerror = () => reject(req.error);
req.onsuccess = () => {
const tx = req.result.transaction('queue', 'readonly');
const all = tx.objectStore('queue').getAll();
all.onsuccess = () =>
resolve(
all.result.filter((r: { status: string }) => r.status === 'blocked').length
);
all.onerror = () => reject(all.error);
};
});
}),
{ timeout: 10_000 }
)
.toBe(1);
});
});

16
frontend/src/lib/ui-store.ts
View File

@@ -7,8 +7,16 @@ export const showBottomNav = writable(true);
// Controls the UploadSheet overlay. FAB sets true; sheet sets false.
export const uploadSheetOpen = writable(false);
// Count of items currently pending or uploading — shown as FAB badge.
export const uploadBadgeCount = derived(
queueItems,
($items) => $items.filter((i) => i.status === 'pending' || i.status === 'uploading').length
// Count of items still needing attention — shown as FAB badge. Includes the terminal
// states on purpose: counting only pending/uploading meant a rejected upload decremented
// the badge exactly as if it had succeeded, so the failure was indistinguishable from a
// completed upload. 'blocked' and 'error' stay counted until the user clears or retries them.
export const uploadBadgeCount = derived(queueItems, ($items) =>
$items.filter(
(i) =>
i.status === 'pending' ||
i.status === 'uploading' ||
i.status === 'error' ||
i.status === 'blocked'
).length
);

6
frontend/src/lib/upload-queue.ts
View File

@@ -3,6 +3,7 @@ import { writable, get } from 'svelte/store';
import { getToken, getUserId, clearAuth } from './auth';
import { onSseEvent } from './sse';
import { refreshQuota } from './quota-store';
import { toast } from './toast-store';
export interface QueueItem {
id: string;
@@ -629,6 +630,11 @@ async function uploadItem(id: string): Promise<void> {
entry.error = e.message;
await database.put(STORE_NAME, entry);
updateItemStatus(id, 'blocked', e.message);
// Tell the user NOW. The queue list only lives on /upload, and the flow sends
// them straight to /feed after staging a photo — so without this a rejected
// upload was silently swallowed: the blob is gone, the FAB badge drops exactly
// as if it had succeeded, and the photo simply never appears.
toast(`${entry.fileName}: ${e.message}`, 'error');
return;
}
const msg = e instanceof Error ? e.message : 'Upload fehlgeschlagen.';

9
frontend/src/routes/upload/+page.svelte
View File

@@ -10,6 +10,7 @@
import { onMount, onDestroy } from 'svelte';
import { quotaStore, refreshQuota } from '$lib/quota-store';
import ConfirmSheet from '$lib/components/ConfirmSheet.svelte';
import UploadQueue from '$lib/components/UploadQueue.svelte';
import IconButton from '$lib/components/IconButton.svelte';
import { vibrate } from '$lib/haptics';
import type { PendingFile } from '$lib/pending-upload-store';
@@ -344,5 +345,13 @@
: 'Hochladen'}
{/if}
</button>
<!--
The queue list. This component existed, complete with the per-item error text, the
"Erneut" retry button and the rate-limit countdown — and was never imported anywhere,
so none of it could be reached. A rejected upload wrote a clear reason into a store
nothing rendered.
-->
<UploadQueue />
</div>
</div>