Content dedup on both sides, stored checksums, spool-free zips, upload stats
ci / docker (push) Successful in 12s
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
This commit is contained in:
+62
-11
@@ -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.
|
||||
|
||||
Reference in New Issue
Block a user