Process slow followers in unicast job queue

2 files changed, 36 insertions, 10 deletions

diff --git a/server/lib/activitypub/send/utils.ts b/server/lib/activitypub/send/utils.ts
index 7cd8030e1..7729703b8 100644
--- a/server/lib/activitypub/send/utils.ts
+++ b/server/lib/activitypub/send/utils.ts
@@ -1,4 +1,5 @@
 import { Transaction } from 'sequelize'
+import { ActorFollowHealthCache } from '@server/lib/actor-follow-health-cache'
 import { getServerActor } from '@server/models/application/application'
 import { ContextType } from '@shared/models/activitypub/context'
 import { Activity, ActivityAudience } from '../../../../shared/models/activitypub'
@@ -119,16 +120,41 @@ async function broadcastToActors (
 function broadcastTo (uris: string[], data: any, byActor: MActorId, contextType?: ContextType) {
   if (uris.length === 0) return undefined
 
-  logger.debug('Creating broadcast job.', { uris })
+  const broadcastUris: string[] = []
+  const unicastUris: string[] = []
 
-  const payload = {
-    uris,
-    signatureActorId: byActor.id,
-    body: data,
-    contextType
+  // Bad URIs could be slow to respond, prefer to process them in a dedicated queue
+  for (const uri of uris) {
+    if (ActorFollowHealthCache.Instance.isBadInbox(uri)) {
+      unicastUris.push(uri)
+    } else {
+      broadcastUris.push(uri)
+    }
+  }
+
+  logger.debug('Creating broadcast job.', { broadcastUris, unicastUris })
+
+  if (broadcastUris.length !== 0) {
+    const payload = {
+      uris: broadcastUris,
+      signatureActorId: byActor.id,
+      body: data,
+      contextType
+    }
+
+    JobQueue.Instance.createJob({ type: 'activitypub-http-broadcast', payload })
   }
 
-  return JobQueue.Instance.createJob({ type: 'activitypub-http-broadcast', payload })
+  for (const unicastUri of unicastUris) {
+    const payload = {
+      uri: unicastUri,
+      signatureActorId: byActor.id,
+      body: data,
+      contextType
+    }
+
+    JobQueue.Instance.createJob({ type: 'activitypub-http-unicast', payload })
+  }
 }
 
 function unicastTo (data: any, byActor: MActorId, toActorUrl: string, contextType?: ContextType) {
diff --git a/server/lib/activitypub/videos/refresh.ts b/server/lib/activitypub/videos/refresh.ts
index 3af08acf4..9f952a218 100644
--- a/server/lib/activitypub/videos/refresh.ts
+++ b/server/lib/activitypub/videos/refresh.ts
@@ -1,10 +1,10 @@
 import { logger, loggerTagsFactory } from '@server/helpers/logger'
 import { PeerTubeRequestError } from '@server/helpers/requests'
-import { ActorFollowScoreCache } from '@server/lib/files-cache'
 import { VideoLoadByUrlType } from '@server/lib/model-loaders'
 import { VideoModel } from '@server/models/video/video'
 import { MVideoAccountLightBlacklistAllFiles, MVideoThumbnail } from '@server/types/models'
 import { HttpStatusCode } from '@shared/models'
+import { ActorFollowHealthCache } from '../../actor-follow-health-cache'
 import { fetchRemoteVideo, SyncParam, syncVideoExternalAttributes } from './shared'
 import { APVideoUpdater } from './updater'
 
@@ -39,7 +39,7 @@ async function refreshVideoIfNeeded (options: {
 
     await syncVideoExternalAttributes(video, videoObject, options.syncParam)
 
-    ActorFollowScoreCache.Instance.addGoodServerId(video.VideoChannel.Actor.serverId)
+    ActorFollowHealthCache.Instance.addGoodServerId(video.VideoChannel.Actor.serverId)
 
     return video
   } catch (err) {
@@ -53,7 +53,7 @@ async function refreshVideoIfNeeded (options: {
 
     logger.warn('Cannot refresh video %s.', options.video.url, { err, ...lTags() })
 
-    ActorFollowScoreCache.Instance.addBadServerId(video.VideoChannel.Actor.serverId)
+    ActorFollowHealthCache.Instance.addBadServerId(video.VideoChannel.Actor.serverId)
 
     // Don't refresh in loop
     await video.setAsRefreshed()