2 Commits
Author SHA1 Message Date
nils 6259ca84d8 Content dedup on both sides, stored checksums, spool-free zips, upload stats
ci / docker (push) Successful in 12s
- sha256+crc32 hashed during upload streaming; unique index per album
- duplicate content returns the existing photo (race-safe via 23505)
- client hashes locally (WebCrypto) and skips the transfer entirely for
  content the album already has
- zip downloads stream S3->response directly using the stored crc32;
  pre-hash photos spool once and self-heal (crc via zip, sha via reprocess)
- upload UI: overall progress bar, bytes, live speed, ETA
2026-07-17 14:41:07 +02:00
nils a6819809a7 Gallery: true justified layout (row-height scaling, no crop, no spacer) 2026-07-17 14:12:57 +02:00
12 changed files with 378 additions and 105 deletions
Generated
+1
View File
@@ -2562,6 +2562,7 @@ dependencies = [
"cookie",
"crc32fast",
"futures",
"hex",
"image",
"rand 0.8.7",
"reqwest",
+1
View File
@@ -25,6 +25,7 @@ axum = { version = "0.8", features = ["macros"] }
axum-extra = { version = "0.10", features = ["cookie", "cookie-signed"] }
chrono = { version = "0.4", features = ["serde"] }
futures = "0.3"
hex = "0.4"
image = "0.25"
rand = "0.8"
reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
+7
View File
@@ -73,6 +73,13 @@ export function postDownload(url, ids = '') {
form.remove()
}
// SHA-256 of a File, matching the server's content hash — used to skip
// uploading bytes the album already has.
export async function sha256Hex(file) {
const digest = await crypto.subtle.digest('SHA-256', await file.arrayBuffer())
return [...new Uint8Array(digest)].map((b) => b.toString(16).padStart(2, '0')).join('')
}
export function uploadFile(url, file, onProgress) {
return new Promise((resolve, reject) => {
const xhr = new XMLHttpRequest()
+76 -32
View File
@@ -1,42 +1,86 @@
import { useEffect, useRef, useState } from 'react'
import { imgUrl } from '../api'
// Justified gallery: rows are built with flexbox, each tile's flex-grow is
// proportional to its aspect ratio so rows fill the container edge to edge.
// When `selected`/`onToggleSelect` are provided, tiles get a select checkmark;
// selection state lives in the parent so it survives lightbox open/close.
// True justified layout: pack photos greedily into rows at their real aspect
// ratios, then scale each row's height so it fills the container width
// exactly. No cropping, no stretch, and the last row simply renders at the
// target height instead of being padded by a spacer.
function layoutRows(photos, containerWidth, targetHeight, gap) {
const rows = []
let row = []
let arSum = 0
let index = 0
for (const photo of photos) {
const ar = photo.width && photo.height ? photo.width / photo.height : 1.5
row.push({ photo, ar, index: index++ })
arSum += ar
const gaps = (row.length - 1) * gap
if (arSum * targetHeight + gaps >= containerWidth) {
rows.push({ items: row, height: (containerWidth - gaps) / arSum })
row = []
arSum = 0
}
}
if (row.length > 0) {
const gaps = (row.length - 1) * gap
rows.push({
items: row,
height: Math.min(targetHeight, (containerWidth - gaps) / arSum),
})
}
return rows
}
export default function Gallery({ photos, onOpen, overlay, selected, onToggleSelect }) {
const containerRef = useRef(null)
const [width, setWidth] = useState(0)
useEffect(() => {
const el = containerRef.current
if (!el) return
const observer = new ResizeObserver((entries) => setWidth(entries[0].contentRect.width))
observer.observe(el)
return () => observer.disconnect()
}, [])
if (photos.length === 0) return null
const gap = 6
const targetHeight = width < 700 ? 170 : 240
const rows = width > 0 ? layoutRows(photos, width, targetHeight, gap) : []
const selecting = selected && selected.size > 0
return (
<div className={`gallery${selecting ? ' selecting' : ''}`}>
{photos.map((p, i) => {
const ar = p.width && p.height ? p.width / p.height : 1.5
const isSelected = selected ? selected.has(p.id) : false
return (
<div
key={p.id}
className={`g-item${isSelected ? ' selected' : ''}`}
style={{ '--ar': ar }}
onClick={() => onOpen && onOpen(i)}
>
<img src={imgUrl(p, 'thumb')} loading="lazy" alt={p.filename} />
{onToggleSelect && (
<button
className="g-check"
title={isSelected ? 'Deselect' : 'Select'}
onClick={(e) => {
e.stopPropagation()
onToggleSelect(p.id)
}}
<div ref={containerRef} className={`gallery${selecting ? ' selecting' : ''}`}>
{rows.map((row) => (
<div key={row.items[0].photo.id} className="g-row" style={{ height: row.height }}>
{row.items.map(({ photo: p, ar, index }) => {
const isSelected = selected ? selected.has(p.id) : false
return (
<div
key={p.id}
className={`g-item${isSelected ? ' selected' : ''}`}
style={{ width: ar * row.height }}
onClick={() => onOpen && onOpen(index)}
>
</button>
)}
{overlay && overlay(p)}
</div>
)
})}
<div className="g-spacer" />
<img src={imgUrl(p, 'thumb')} loading="lazy" alt={p.filename} />
{onToggleSelect && (
<button
className="g-check"
title={isSelected ? 'Deselect' : 'Select'}
onClick={(e) => {
e.stopPropagation()
onToggleSelect(p.id)
}}
>
</button>
)}
{overlay && overlay(p)}
</div>
)
})}
</div>
))}
</div>
)
}
+79 -19
View File
@@ -1,21 +1,57 @@
import { useCallback, useEffect, useRef, useState } from 'react'
import { Link, useNavigate, useParams } from 'react-router-dom'
import { api, postDownload, uploadFile } from '../api'
import { api, postDownload, sha256Hex, uploadFile } from '../api'
import Gallery from '../components/Gallery'
import Lightbox from '../components/Lightbox'
import SelectionBar from '../components/SelectionBar'
import SelectionBar, { fmtBytes } from '../components/SelectionBar'
import Stars from '../components/Stars'
import useSelection from '../useSelection'
function fmtEta(seconds) {
if (!isFinite(seconds) || seconds < 0) return ''
if (seconds < 60) return `${Math.ceil(seconds)}s`
if (seconds < 3600) return `${Math.ceil(seconds / 60)} min`
return `${Math.floor(seconds / 3600)}h ${Math.ceil((seconds % 3600) / 60)} min`
}
const UPLOAD_CONCURRENCY = 3
function UploadZone({ albumId, onUploaded }) {
const [queue, setQueue] = useState([])
const [dragging, setDragging] = useState(false)
const [speed, setSpeed] = useState(0)
const inputRef = useRef(null)
const running = useRef(0)
const pending = useRef([])
const lastRefresh = useRef(0)
const loadedRef = useRef(0)
const totalBytes = queue.reduce((sum, item) => sum + item.file.size, 0)
const loadedBytes = queue.reduce(
(sum, item) =>
sum + (item.status === 'done' ? item.file.size : (item.progress || 0) * item.file.size),
0,
)
loadedRef.current = loadedBytes
const active = queue.some((item) =>
['uploading', 'queued', 'checking'].includes(item.status),
)
// Sample throughput once a second (EMA-smoothed) while uploads run.
useEffect(() => {
if (!active) {
setSpeed(0)
return
}
let last = { loaded: loadedRef.current, time: Date.now() }
const timer = setInterval(() => {
const now = Date.now()
const instant = (loadedRef.current - last.loaded) / ((now - last.time) / 1000)
last = { loaded: loadedRef.current, time: now }
setSpeed((prev) => (prev > 0 ? prev * 0.7 + instant * 0.3 : instant))
}, 1000)
return () => clearInterval(timer)
}, [active])
// Refresh the album at most every 5s during a bulk upload (the processing
// poll keeps it fresh anyway), plus once when the queue drains.
@@ -31,23 +67,27 @@ function UploadZone({ albumId, onUploaded }) {
while (running.current < UPLOAD_CONCURRENCY && pending.current.length > 0) {
const item = pending.current.shift()
running.current += 1
setQueue((q) =>
q.map((x) => (x.key === item.key ? { ...x, status: 'uploading' } : x)),
)
const url = `/api/albums/${albumId}/photos?filename=${encodeURIComponent(item.file.name)}`
uploadFile(url, item.file, (p) =>
setQueue((q) => q.map((x) => (x.key === item.key ? { ...x, progress: p } : x))),
)
.then(() =>
setQueue((q) =>
q.map((x) => (x.key === item.key ? { ...x, status: 'done', progress: 1 } : x)),
),
)
.catch((e) =>
setQueue((q) =>
q.map((x) => (x.key === item.key ? { ...x, status: 'error', error: e.message } : x)),
),
)
const update = (patch) =>
setQueue((q) => q.map((x) => (x.key === item.key ? { ...x, ...patch } : x)))
const transfer = async () => {
// Hash locally first: content the album already has is skipped
// without transferring a single byte.
update({ status: 'checking' })
try {
const hash = await sha256Hex(item.file)
await api(`/api/albums/${albumId}/photos/by-hash/${hash}`)
update({ status: 'skipped', progress: 1 })
return
} catch {
// 404 (not there yet) or hashing unavailable — upload normally.
}
update({ status: 'uploading' })
const url = `/api/albums/${albumId}/photos?filename=${encodeURIComponent(item.file.name)}`
await uploadFile(url, item.file, (p) => update({ progress: p }))
update({ status: 'done', progress: 1 })
}
transfer()
.catch((e) => update({ status: 'error', error: e.message }))
.finally(() => {
running.current -= 1
refresh()
@@ -95,6 +135,22 @@ function UploadZone({ albumId, onUploaded }) {
}}
/>
<p>Drop RAWs or JPGs here, or click to select</p>
{queue.length > 0 && (
<div className="upload-summary" onClick={(e) => e.stopPropagation()}>
<progress className="upload-total" value={loadedBytes} max={totalBytes || 1} />
<span className="muted">
{queue.filter((i) => i.status === 'done' || i.status === 'skipped').length} /{' '}
{queue.length} files ·{' '}
{fmtBytes(loadedBytes)} of {fmtBytes(totalBytes)}
{active && speed > 0 && (
<>
{' · '}
{fmtBytes(speed)}/s · ~{fmtEta((totalBytes - loadedBytes) / speed)} left
</>
)}
</span>
</div>
)}
{queue.length > 0 && (
<ul className="upload-list" onClick={(e) => e.stopPropagation()}>
{queue.map((item) => (
@@ -102,6 +158,10 @@ function UploadZone({ albumId, onUploaded }) {
<span className="upload-name">{item.file.name}</span>
{item.status === 'error' ? (
<span className="error">{item.error}</span>
) : item.status === 'skipped' ? (
<span className="muted">already uploaded</span>
) : item.status === 'checking' ? (
<span className="muted">checking</span>
) : (
<progress value={item.progress} max="1" />
)}
+18 -13
View File
@@ -191,15 +191,17 @@ input:focus {
/* justified gallery */
.gallery {
display: flex;
flex-wrap: wrap;
flex-direction: column;
gap: 6px;
margin: 1rem 0;
}
.g-row {
display: flex;
gap: 6px;
}
.g-item {
position: relative;
height: 240px;
flex-grow: calc(var(--ar) * 100);
flex-basis: calc(var(--ar) * 240px);
flex: none;
border-radius: 4px;
overflow: hidden;
cursor: pointer;
@@ -211,11 +213,6 @@ input:focus {
object-fit: cover;
display: block;
}
.g-spacer {
flex-grow: 1000000;
flex-basis: 0;
height: 0;
}
.g-overlay {
position: absolute;
bottom: 0;
@@ -310,6 +307,18 @@ input:focus {
.upload-zone p {
margin: 0;
}
.upload-summary {
display: flex;
flex-direction: column;
gap: 0.3rem;
margin-top: 1rem;
cursor: default;
font-size: 0.9rem;
}
.upload-total {
width: 100%;
accent-color: var(--accent);
}
.upload-list {
list-style: none;
margin: 1rem 0 0;
@@ -556,10 +565,6 @@ progress {
}
@media (max-width: 700px) {
.g-item {
height: 160px;
flex-basis: calc(var(--ar) * 160px);
}
.lb-stage {
padding: 0 0.5rem;
}
+10
View File
@@ -0,0 +1,10 @@
-- Content hashes: sha256 powers duplicate-upload detection (same content in
-- the same album is returned instead of copied); crc32 lets zip downloads
-- stream originals straight from S3 without a local spool pass.
-- Both are null for photos uploaded before this migration; they self-heal on
-- reprocess (sha256 + crc32) and on zip download (crc32).
alter table photos add column sha256 text;
alter table photos add column crc32 bigint;
create unique index photos_album_sha_uidx on photos (album_id, sha256)
where sha256 is not null;
+37 -4
View File
@@ -62,7 +62,24 @@ pub async fn process_photo(state: &AppState, photo_id: Uuid) -> anyhow::Result<(
.unwrap_or("bin")
.to_lowercase();
let src_path = dir.path().join(format!("original.{extension}"));
download(state, &s3::original_key(photo_id, &photo.filename), &src_path).await?;
let (sha256, crc32) =
download(state, &s3::original_key(photo_id, &photo.filename), &src_path).await?;
if photo.sha256.is_none() || photo.crc32.is_none() {
// Best-effort backfill for pre-hash photos; a duplicate in the same
// album trips the unique index, which is fine — skip silently.
if let Err(e) = sqlx::query(
"update photos set sha256 = coalesce(sha256, $2), crc32 = coalesce(crc32, $3)
where id = $1",
)
.bind(photo_id)
.bind(&sha256)
.bind(crc32)
.execute(&state.db)
.await
{
tracing::debug!("hash backfill skipped for {photo_id}: {e}");
}
}
let meta = exif_metadata(&src_path).await?;
@@ -103,7 +120,11 @@ pub async fn process_photo(state: &AppState, photo_id: Uuid) -> anyhow::Result<(
Ok(())
}
async fn download(state: &AppState, key: &str, path: &Path) -> anyhow::Result<()> {
/// Download the original, hashing along the way so legacy photos (uploaded
/// before hashes existed) can be backfilled.
async fn download(state: &AppState, key: &str, path: &Path) -> anyhow::Result<(String, i64)> {
use sha2::Digest;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let object = state
.s3
.get_object()
@@ -114,8 +135,20 @@ async fn download(state: &AppState, key: &str, path: &Path) -> anyhow::Result<()
.with_context(|| format!("fetching s3://{}/{key}", state.config.s3_bucket))?;
let mut reader = object.body.into_async_read();
let mut file = tokio::fs::File::create(path).await?;
tokio::io::copy(&mut reader, &mut file).await?;
Ok(())
let mut sha = sha2::Sha256::new();
let mut crc = crc32fast::Hasher::new();
let mut buf = vec![0u8; 128 * 1024];
loop {
let n = reader.read(&mut buf).await?;
if n == 0 {
break;
}
sha.update(&buf[..n]);
crc.update(&buf[..n]);
file.write_all(&buf[..n]).await?;
}
file.flush().await?;
Ok((hex::encode(sha.finalize()), i64::from(crc.finalize())))
}
#[derive(Default)]
+4
View File
@@ -115,6 +115,10 @@ pub struct Photo {
pub height: Option<i32>,
pub taken_at: Option<DateTime<Utc>>,
pub processed_at: Option<DateTime<Utc>>,
#[serde(skip_serializing)]
pub sha256: Option<String>,
#[serde(skip_serializing)]
pub crc32: Option<i64>,
pub created_at: DateTime<Utc>,
}
+1
View File
@@ -77,6 +77,7 @@ pub fn router(state: &AppState) -> Router<AppState> {
get(shares::list).post(shares::create),
)
.route("/api/albums/{id}/zip", post(zip::album_zip))
.route("/api/albums/{id}/photos/by-hash/{sha256}", get(photos::by_hash))
.route("/api/photos/{id}", delete(photos::delete))
.route("/api/photos/{id}/reprocess", post(photos::reprocess))
.route("/api/shares/{id}", delete(shares::delete))
+82 -26
View File
@@ -5,6 +5,7 @@ use axum::http::HeaderMap;
use axum::Json;
use futures::StreamExt;
use serde::Deserialize;
use sha2::Digest;
use tokio::io::AsyncWriteExt;
use uuid::Uuid;
@@ -60,17 +61,23 @@ pub async fn upload(
.unwrap_or("application/octet-stream")
.to_string();
// Stream the request body to a temp file so large raws never sit in memory.
// Stream the request body to a temp file so large raws never sit in
// memory, hashing as it flows: sha256 for duplicate detection, crc32 for
// spool-free zip downloads.
let dir = tempfile::tempdir().map_err(anyhow::Error::from)?;
let path = dir.path().join("upload.bin");
let mut file = tokio::fs::File::create(&path)
.await
.map_err(anyhow::Error::from)?;
let mut stream = body.into_data_stream();
let mut sha = sha2::Sha256::new();
let mut crc = crc32fast::Hasher::new();
let mut size: i64 = 0;
while let Some(chunk) = stream.next().await {
let chunk = chunk.map_err(|e| ApiError::bad_request(format!("upload aborted: {e}")))?;
size += chunk.len() as i64;
sha.update(&chunk);
crc.update(&chunk);
file.write_all(&chunk).await.map_err(anyhow::Error::from)?;
}
file.flush().await.map_err(anyhow::Error::from)?;
@@ -78,6 +85,20 @@ pub async fn upload(
if size == 0 {
return Err(ApiError::bad_request("empty upload"));
}
let sha256 = hex::encode(sha.finalize());
let crc32 = i64::from(crc.finalize());
// Same content already in this album? Return it — re-dragging a folder
// after a partial upload just fills the gaps instead of duplicating.
let existing: Option<Photo> =
sqlx::query_as("select * from photos where album_id = $1 and sha256 = $2")
.bind(album_id)
.bind(&sha256)
.fetch_optional(&state.db)
.await?;
if let Some(existing) = existing {
return Ok(Json(existing));
}
// Upload to S3 first, then create the row and enqueue processing in one
// transaction — a photo row can never exist without its job, and a failed
@@ -86,30 +107,33 @@ pub async fn upload(
let key = s3::original_key(photo_id, &filename);
s3::put_file(&state, &key, &path, &content_type).await?;
let result: ApiResult<(sqlx::Transaction<'static, sqlx::Postgres>, Photo)> = async {
let mut tx = state.db.begin().await?;
let photo: Photo = sqlx::query_as(
"insert into photos (id, album_id, filename, content_type, size_bytes, status)
values ($1, $2, $3, $4, $5, $6)
returning *",
)
.bind(photo_id)
.bind(album_id)
.bind(&filename)
.bind(&content_type)
.bind(size)
.bind(PhotoStatus::Uploaded.as_str())
.fetch_one(&mut *tx)
.await?;
jobs::enqueue(
&mut *tx,
JobKind::ProcessPhoto,
serde_json::json!({ "photo_id": photo_id }),
)
.await?;
Ok((tx, photo))
}
.await;
let result: Result<(sqlx::Transaction<'static, sqlx::Postgres>, Photo), sqlx::Error> =
async {
let mut tx = state.db.begin().await?;
let photo: Photo = sqlx::query_as(
"insert into photos (id, album_id, filename, content_type, size_bytes, status, sha256, crc32)
values ($1, $2, $3, $4, $5, $6, $7, $8)
returning *",
)
.bind(photo_id)
.bind(album_id)
.bind(&filename)
.bind(&content_type)
.bind(size)
.bind(PhotoStatus::Uploaded.as_str())
.bind(&sha256)
.bind(crc32)
.fetch_one(&mut *tx)
.await?;
jobs::enqueue(
&mut *tx,
JobKind::ProcessPhoto,
serde_json::json!({ "photo_id": photo_id }),
)
.await?;
Ok((tx, photo))
}
.await;
let (tx, photo) = match result {
Ok(pair) => pair,
@@ -118,7 +142,24 @@ pub async fn upload(
if let Err(cleanup) = s3::delete_prefix(&state, &s3::photo_prefix(photo_id)).await {
tracing::error!("failed to clean up s3 after aborted upload: {cleanup:#}");
}
return Err(e);
// Concurrent identical upload beat us to the unique index — hand
// back the winner instead of an error.
let unique_violation = matches!(
&e,
sqlx::Error::Database(db) if db.code().as_deref() == Some("23505")
);
if unique_violation {
let winner: Option<Photo> =
sqlx::query_as("select * from photos where album_id = $1 and sha256 = $2")
.bind(album_id)
.bind(&sha256)
.fetch_optional(&state.db)
.await?;
if let Some(winner) = winner {
return Ok(Json(winner));
}
}
return Err(e.into());
}
};
if let Err(e) = tx.commit().await {
@@ -133,6 +174,21 @@ pub async fn upload(
Ok(Json(photo))
}
/// Client-side dedup support: lets the uploader skip transferring files whose
/// content already exists in the album.
pub async fn by_hash(
State(state): State<AppState>,
Path((album_id, sha256)): Path<(Uuid, String)>,
) -> ApiResult<Json<Photo>> {
let photo: Option<Photo> =
sqlx::query_as("select * from photos where album_id = $1 and sha256 = $2")
.bind(album_id)
.bind(&sha256)
.fetch_optional(&state.db)
.await?;
photo.map(Json).ok_or_else(ApiError::not_found)
}
pub async fn delete(
State(state): State<AppState>,
Path(photo_id): Path<Uuid>,
+62 -11
View File
@@ -136,6 +136,9 @@ struct Entry {
offset: u64,
dos_time: u16,
dos_date: u16,
/// Stored at upload/processing time; photos from before hashes existed
/// have None and take the slower spool path (which backfills it).
crc: Option<u32>,
}
struct ZipPlan {
@@ -176,6 +179,7 @@ fn plan_zip(photos: &[Photo]) -> Result<ZipPlan, ApiError> {
offset: entry_offset,
dos_time,
dos_date,
crc: photo.crc32.map(|v| v as u32),
});
}
let cd_offset = offset;
@@ -277,9 +281,23 @@ fn stream_zip(state: AppState, photos: Vec<Photo>, album_name: &str) -> ApiResul
.map_err(|e| anyhow::anyhow!("building response: {e}").into())
}
/// Download an original into an anonymous temp file, computing its CRC-32 and
/// verifying the byte count matches what the zip plan promised.
fn spool(state: &AppState, key: String, expected_size: u64) -> JoinHandle<anyhow::Result<(tokio::fs::File, u32)>> {
enum Fetched {
/// CRC already known — the body streams straight into the response.
Direct(Box<aws_sdk_s3::operation::get_object::GetObjectOutput>),
/// Pre-hash photo: spooled to a temp file to compute the CRC first.
Spooled(tokio::fs::File, u32),
}
/// Start fetching an original. With a known CRC this only opens the S3
/// response (the body is consumed later, straight into the zip stream);
/// otherwise the object is spooled to an anonymous temp file to compute the
/// CRC, verifying the byte count the zip plan promised.
fn fetch_entry(
state: &AppState,
key: String,
expected_size: u64,
crc_known: bool,
) -> JoinHandle<anyhow::Result<Fetched>> {
let state = state.clone();
tokio::spawn(async move {
let object = state
@@ -290,6 +308,9 @@ fn spool(state: &AppState, key: String, expected_size: u64) -> JoinHandle<anyhow
.send()
.await
.map_err(|e| anyhow::anyhow!("fetching {key}: {e}"))?;
if crc_known {
return Ok(Fetched::Direct(Box::new(object)));
}
let mut file = tokio::fs::File::from_std(tempfile::tempfile()?);
let mut reader = object.body.into_async_read();
let mut hasher = crc32fast::Hasher::new();
@@ -310,7 +331,7 @@ fn spool(state: &AppState, key: String, expected_size: u64) -> JoinHandle<anyhow
);
file.flush().await?;
file.seek(std::io::SeekFrom::Start(0)).await?;
Ok((file, hasher.finalize()))
Ok(Fetched::Spooled(file, hasher.finalize()))
})
}
@@ -323,20 +344,29 @@ async fn write_zip(
const FLAGS: u16 = 0x0800;
let mut crcs = Vec::with_capacity(plan.entries.len());
// Prefetch: spool the next object from S3 while streaming the current one.
let mut pending: Option<JoinHandle<anyhow::Result<(tokio::fs::File, u32)>>> = None;
// Prefetch: start fetching the next object while streaming the current one.
let mut pending: Option<JoinHandle<anyhow::Result<Fetched>>> = None;
for (i, entry) in plan.entries.iter().enumerate() {
let current = match pending.take() {
Some(handle) => handle,
None => spool(state, entry.s3_key.clone(), entry.size),
None => fetch_entry(state, entry.s3_key.clone(), entry.size, entry.crc.is_some()),
};
if let Some(next) = plan.entries.get(i + 1) {
pending = Some(spool(state, next.s3_key.clone(), next.size));
pending = Some(fetch_entry(
state,
next.s3_key.clone(),
next.size,
next.crc.is_some(),
));
}
let (mut file, crc) = current
let fetched = current
.await
.map_err(|e| anyhow::anyhow!("spool task failed: {e}"))?
.map_err(|e| anyhow::anyhow!("fetch task failed: {e}"))?
.map_err(|e| anyhow::anyhow!("photo {}: {e:#}", entry.photo_id))?;
let crc = match &fetched {
Fetched::Direct(_) => entry.crc.expect("direct fetch implies known crc"),
Fetched::Spooled(_, crc) => *crc,
};
crcs.push(crc);
let mut lfh = Vec::with_capacity(30 + entry.name.len());
@@ -353,7 +383,28 @@ async fn write_zip(
lfh.extend_from_slice(&0u16.to_le_bytes()); // extra len
lfh.extend_from_slice(&entry.name);
out.write_all(&lfh).await?;
tokio::io::copy(&mut file, &mut out).await?;
match fetched {
Fetched::Direct(object) => {
let mut reader = object.body.into_async_read();
let copied = tokio::io::copy(&mut reader, &mut out).await?;
anyhow::ensure!(
copied == entry.size,
"{} is {copied} bytes in s3 but {} in the database",
entry.s3_key,
entry.size
);
}
Fetched::Spooled(mut file, crc) => {
tokio::io::copy(&mut file, &mut out).await?;
// Self-heal: store the freshly computed crc so the next
// download of this photo streams directly.
let _ = sqlx::query("update photos set crc32 = coalesce(crc32, $2) where id = $1")
.bind(entry.photo_id)
.bind(i64::from(crc))
.execute(&state.db)
.await;
}
}
}
// Central directory.