LemmyNet/lemmy
2
0
Fork 0
mirror of https://github.com/LemmyNet/lemmy.git synced 2024-11-16 09:24:00 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Fix resending activities (fixes #1282) (#2109)

Browse source
This commit is contained in:
Nutomic 2022-03-01 18:40:05 +00:00 committed by GitHub
parent fb4c47d602
commit 63565712ad
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 25 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

41
crates/apub_lib/src/activity_queue.rs
View file

@ -1,5 +1,5 @@
use crate::{signatures::sign_and_send, traits::ActorType}; use crate::{signatures::sign_and_send, traits::ActorType};
use anyhow::{Context, Error}; use anyhow::{anyhow, Context, Error};
use background_jobs::{ use background_jobs::{
memory_storage::Storage, memory_storage::Storage,
ActixJob, ActixJob,
@ -33,7 +33,13 @@ pub async fn send_activity(
private_key: actor.private_key().context(location_info!())?, private_key: actor.private_key().context(location_info!())?,
}; };
if env::var("APUB_TESTING_SEND_SYNC").is_ok() { if env::var("APUB_TESTING_SEND_SYNC").is_ok() {
do_send(message, client).await?; let res = do_send(message, client).await;
// Don't fail on error, as we intentionally do some invalid actions in tests, to verify that
// they are rejected on the receiving side. These errors shouldn't bubble up to make the API
// call fail. This matches the behaviour in production.
if let Err(e) = res {
warn!("{}", e);
}
} else { } else {
activity_queue.queue::<SendActivityTask>(message).await?; activity_queue.queue::<SendActivityTask>(message).await?;
} }
@ -77,27 +83,32 @@ async fn do_send(task: SendActivityTask, client: &ClientWithMiddleware) -> Resul
) )
.await; .await;
match result { let r: Result<(), Error> = match result {
Ok(o) => { Ok(o) => {
if !o.status().is_success() { if !o.status().is_success() {
let status = o.status(); let status = o.status();
let text = o.text().await?; let text = o.text().await?;
warn!( Err(anyhow!(
"Send {} to {} failed with status {}: {}", "Send {} to {} failed with status {}: {}",
task.activity_id, task.inbox, status, text, task.activity_id,
); task.inbox,
} status,
} text,
Err(e) => { ))
warn!( } else {
"Failed to send activity {} to {}: {}",
&task.activity_id, task.inbox, e
);
}
}
Ok(()) Ok(())
} }
}
Err(e) => Err(anyhow!(
"Failed to send activity {} to {}: {}",
&task.activity_id,
task.inbox,
e
)),
};
r
}
pub fn create_activity_queue(client: ClientWithMiddleware) -> Manager { pub fn create_activity_queue(client: ClientWithMiddleware) -> Manager {
// Configure and start our workers // Configure and start our workers