mirrors/pict-rs
1
0
Fork 0
mirror of https://git.asonix.dog/asonix/pict-rs.git synced 2024-11-28 20:41:00 +00:00
Code Issues Projects Releases Packages Wiki Activity

Fix dangling unprocessed uploads

Browse source 
Adds error boundary around backgrounded ingest
This commit is contained in:
asonix 2023-07-17 22:30:10 -05:00
parent 42c801f0fe
commit 558605381d
1 changed files with 20 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

24
src/queue/process.rs
View file

@ -1,5 +1,5 @@
use crate::{ use crate::{
error::Error, error::{Error, UploadError},
formats::InputProcessableFormat, formats::InputProcessableFormat,
ingest::Session, ingest::Session,
queue::{Base64Bytes, LocalBoxFuture, Process}, queue::{Base64Bytes, LocalBoxFuture, Process},
@ -71,27 +71,37 @@ async fn process_ingest<R, S>(
unprocessed_identifier: Vec<u8>, unprocessed_identifier: Vec<u8>,
upload_id: UploadId, upload_id: UploadId,
declared_alias: Option<Alias>, declared_alias: Option<Alias>,
media: &crate::config::Media, media: &'static crate::config::Media,
) -> Result<(), Error> ) -> Result<(), Error>
where where
R: FullRepo + 'static, R: FullRepo + 'static,
S: Store, S: Store + 'static,
{ {
let fut = async { let fut = async {
let unprocessed_identifier = S::Identifier::from_bytes(unprocessed_identifier)?; let unprocessed_identifier = S::Identifier::from_bytes(unprocessed_identifier)?;
let stream = store let ident = unprocessed_identifier.clone();
.to_stream(&unprocessed_identifier, None, None) let store2 = store.clone();
let repo = repo.clone();
let error_boundary = actix_rt::spawn(async move {
let stream = store2
.to_stream(&ident, None, None)
.await? .await?
.map_err(Error::from); .map_err(Error::from);
let session = crate::ingest::ingest(repo, store, stream, declared_alias, media).await?; let session =
crate::ingest::ingest(&repo, &store2, stream, declared_alias, media).await?;
let token = session.delete_token().await?; let token = session.delete_token().await?;
Ok((session, token)) as Result<(Session<R, S>, DeleteToken), Error>
})
.await;
store.remove(&unprocessed_identifier).await?; store.remove(&unprocessed_identifier).await?;
Ok((session, token)) as Result<(Session<R, S>, DeleteToken), Error> error_boundary.map_err(|_| UploadError::Canceled)?
}; };
let result = match fut.await { let result = match fut.await {