grishka/lemmy
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

make activity channel infallible (#4295)

Browse source
This commit is contained in:
phiresky 2023-12-18 19:17:10 +01:00 committed by GitHub
parent dcb89f52d6
commit 6790b54d4d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 7 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
crates/api_common/src/send_activity.rs
View file

@ -98,9 +98,9 @@ impl ActivityChannel {
Ok(())
}
pub async fn close(outgoing_activities_task: JoinHandle<LemmyResult<()>>) -> LemmyResult<()> {
pub async fn close(outgoing_activities_task: JoinHandle<()>) -> LemmyResult<()> {
ACTIVITY_CHANNEL.keepalive_sender.lock().await.take();
outgoing_activities_task.await??;
outgoing_activities_task.await?;
Ok(())
}
}

7
crates/apub/src/activities/mod.rs
View file

@ -225,11 +225,12 @@ where
Ok(())
}
pub async fn handle_outgoing_activities(context: Data<LemmyContext>) -> LemmyResult<()> {
pub async fn handle_outgoing_activities(context: Data<LemmyContext>) {
while let Some(data) = ActivityChannel::retrieve_activity().await {
match_outgoing_activities(data, &context.reset_request_count()).await?
if let Err(e) = match_outgoing_activities(data, &context.reset_request_count()).await {
tracing::warn!("error while saving outgoing activity to db: {e}");
}
}
Ok(())
}
pub async fn match_outgoing_activities(

1
crates/federate/src/worker.rs
View file

@ -171,6 +171,7 @@ impl InstanceWorker {
.await
.context("failed reading activity from db")?
else {
tracing::debug!("{}: {:?} does not exist", self.instance.domain, id);
self.state.last_successful_id = Some(id);
continue;
};