Revert "Retry activity send in case of timeout or rate limit (#102)"
This reverts commit 5402bc9c19
.
This commit is contained in:
parent
16844f048a
commit
32da1b747c
3 changed files with 13 additions and 68 deletions
|
@ -12,17 +12,14 @@ use crate::{
|
||||||
};
|
};
|
||||||
use bytes::Bytes;
|
use bytes::Bytes;
|
||||||
use futures::StreamExt;
|
use futures::StreamExt;
|
||||||
use http::StatusCode;
|
|
||||||
use httpdate::fmt_http_date;
|
use httpdate::fmt_http_date;
|
||||||
use itertools::Itertools;
|
use itertools::Itertools;
|
||||||
use openssl::pkey::{PKey, Private};
|
use openssl::pkey::{PKey, Private};
|
||||||
use reqwest::{
|
use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
|
||||||
header::{HeaderMap, HeaderName, HeaderValue},
|
|
||||||
Response,
|
|
||||||
};
|
|
||||||
use reqwest_middleware::ClientWithMiddleware;
|
use reqwest_middleware::ClientWithMiddleware;
|
||||||
use serde::Serialize;
|
use serde::Serialize;
|
||||||
use std::{
|
use std::{
|
||||||
|
self,
|
||||||
fmt::{Debug, Display},
|
fmt::{Debug, Display},
|
||||||
time::{Duration, SystemTime},
|
time::{Duration, SystemTime},
|
||||||
};
|
};
|
||||||
|
@ -93,30 +90,20 @@ impl SendActivityTask {
|
||||||
)
|
)
|
||||||
.await?;
|
.await?;
|
||||||
let response = client.execute(request).await?;
|
let response = client.execute(request).await?;
|
||||||
self.handle_response(response).await
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Based on the HTTP status code determines if an activity was delivered successfully. In that case
|
match response {
|
||||||
/// Ok is returned. Otherwise it returns Err and the activity send should be retried later.
|
o if o.status().is_success() => {
|
||||||
///
|
|
||||||
/// Equivalent code in mastodon: https://github.com/mastodon/mastodon/blob/v4.2.8/app/helpers/jsonld_helper.rb#L215-L217
|
|
||||||
async fn handle_response(&self, response: Response) -> Result<(), Error> {
|
|
||||||
match response.status() {
|
|
||||||
status if status.is_success() => {
|
|
||||||
debug!("Activity {self} delivered successfully");
|
debug!("Activity {self} delivered successfully");
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
status
|
o if o.status().is_client_error() => {
|
||||||
if status.is_client_error()
|
let text = o.text_limited().await?;
|
||||||
&& status != StatusCode::REQUEST_TIMEOUT
|
|
||||||
&& status != StatusCode::TOO_MANY_REQUESTS =>
|
|
||||||
{
|
|
||||||
let text = response.text_limited().await?;
|
|
||||||
debug!("Activity {self} was rejected, aborting: {text}");
|
debug!("Activity {self} was rejected, aborting: {text}");
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
status => {
|
o => {
|
||||||
let text = response.text_limited().await?;
|
let status = o.status();
|
||||||
|
let text = o.text_limited().await?;
|
||||||
|
|
||||||
Err(Error::Other(format!(
|
Err(Error::Other(format!(
|
||||||
"Activity {self} failure with status {status}: {text}",
|
"Activity {self} failure with status {status}: {text}",
|
||||||
|
@ -228,6 +215,8 @@ pub(crate) fn generate_request_headers(inbox_url: &Url) -> HeaderMap {
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
use crate::{config::FederationConfig, http_signatures::generate_actor_keypair};
|
use crate::{config::FederationConfig, http_signatures::generate_actor_keypair};
|
||||||
|
use bytes::Bytes;
|
||||||
|
use http::StatusCode;
|
||||||
use std::{
|
use std::{
|
||||||
sync::{atomic::AtomicUsize, Arc},
|
sync::{atomic::AtomicUsize, Arc},
|
||||||
time::Instant,
|
time::Instant,
|
||||||
|
@ -301,48 +290,4 @@ mod tests {
|
||||||
info!("Queue Sent: {:?}", start.elapsed());
|
info!("Queue Sent: {:?}", start.elapsed());
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
|
||||||
async fn test_handle_response() {
|
|
||||||
let keypair = generate_actor_keypair().unwrap();
|
|
||||||
let message = SendActivityTask {
|
|
||||||
actor_id: "http://localhost:8001".parse().unwrap(),
|
|
||||||
activity_id: "http://localhost:8001/activity".parse().unwrap(),
|
|
||||||
activity: "{}".into(),
|
|
||||||
inbox: "http://localhost:8001".parse().unwrap(),
|
|
||||||
private_key: keypair.private_key().unwrap(),
|
|
||||||
http_signature_compat: true,
|
|
||||||
};
|
|
||||||
|
|
||||||
let res = |status| {
|
|
||||||
http::Response::builder()
|
|
||||||
.status(status)
|
|
||||||
.body(vec![])
|
|
||||||
.unwrap()
|
|
||||||
.into()
|
|
||||||
};
|
|
||||||
|
|
||||||
assert!(message.handle_response(res(StatusCode::OK)).await.is_ok());
|
|
||||||
assert!(message
|
|
||||||
.handle_response(res(StatusCode::BAD_REQUEST))
|
|
||||||
.await
|
|
||||||
.is_ok());
|
|
||||||
|
|
||||||
assert!(message
|
|
||||||
.handle_response(res(StatusCode::MOVED_PERMANENTLY))
|
|
||||||
.await
|
|
||||||
.is_err());
|
|
||||||
assert!(message
|
|
||||||
.handle_response(res(StatusCode::REQUEST_TIMEOUT))
|
|
||||||
.await
|
|
||||||
.is_err());
|
|
||||||
assert!(message
|
|
||||||
.handle_response(res(StatusCode::TOO_MANY_REQUESTS))
|
|
||||||
.await
|
|
||||||
.is_err());
|
|
||||||
assert!(message
|
|
||||||
.handle_response(res(StatusCode::INTERNAL_SERVER_ERROR))
|
|
||||||
.await
|
|
||||||
.is_err());
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
|
@ -347,7 +347,7 @@ const _IMPL_DIESEL_NEW_TYPE_FOR_OBJECT_ID: () = {
|
||||||
#[allow(clippy::unwrap_used)]
|
#[allow(clippy::unwrap_used)]
|
||||||
pub mod tests {
|
pub mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
use crate::traits::tests::DbUser;
|
use crate::{fetch::object_id::should_refetch_object, traits::tests::DbUser};
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_deserialize() {
|
fn test_deserialize() {
|
||||||
|
|
|
@ -343,7 +343,7 @@ pub mod tests {
|
||||||
error::Error,
|
error::Error,
|
||||||
fetch::object_id::ObjectId,
|
fetch::object_id::ObjectId,
|
||||||
http_signatures::{generate_actor_keypair, Keypair},
|
http_signatures::{generate_actor_keypair, Keypair},
|
||||||
protocol::verification::verify_domains_match,
|
protocol::{public_key::PublicKey, verification::verify_domains_match},
|
||||||
};
|
};
|
||||||
use activitystreams_kinds::{activity::FollowType, actor::PersonType};
|
use activitystreams_kinds::{activity::FollowType, actor::PersonType};
|
||||||
use once_cell::sync::Lazy;
|
use once_cell::sync::Lazy;
|
||||||
|
|
Loading…
Reference in a new issue