aboutsummaryrefslogtreecommitdiffhomepage
path: root/server/lib/job-queue/handlers/activitypub-http-broadcast.ts
diff options
context:
space:
mode:
authorChocobozzz <me@florianbigard.com>2021-10-13 11:47:32 +0200
committerChocobozzz <me@florianbigard.com>2021-10-13 11:47:32 +0200
commit9db437c8155f3563a33e22ed2896072a9f1fbdb0 (patch)
tree716078fbe1506e0b0d19936f4939e9c530b3b8ab /server/lib/job-queue/handlers/activitypub-http-broadcast.ts
parente81f6ccf989d4573b59ec7b2bf2812fe3e9fb534 (diff)
downloadPeerTube-9db437c8155f3563a33e22ed2896072a9f1fbdb0.tar.gz
PeerTube-9db437c8155f3563a33e22ed2896072a9f1fbdb0.tar.zst
PeerTube-9db437c8155f3563a33e22ed2896072a9f1fbdb0.zip
Process slow followers in unicast job queue
Diffstat (limited to 'server/lib/job-queue/handlers/activitypub-http-broadcast.ts')
-rw-r--r--server/lib/job-queue/handlers/activitypub-http-broadcast.ts16
1 files changed, 10 insertions, 6 deletions
diff --git a/server/lib/job-queue/handlers/activitypub-http-broadcast.ts b/server/lib/job-queue/handlers/activitypub-http-broadcast.ts
index 9b0bb6574..fbf01d276 100644
--- a/server/lib/job-queue/handlers/activitypub-http-broadcast.ts
+++ b/server/lib/job-queue/handlers/activitypub-http-broadcast.ts
@@ -1,10 +1,10 @@
1import { map } from 'bluebird' 1import { map } from 'bluebird'
2import { Job } from 'bull' 2import { Job } from 'bull'
3import { ActorFollowHealthCache } from '@server/lib/actor-follow-health-cache'
3import { ActivitypubHttpBroadcastPayload } from '@shared/models' 4import { ActivitypubHttpBroadcastPayload } from '@shared/models'
4import { logger } from '../../../helpers/logger' 5import { logger } from '../../../helpers/logger'
5import { doRequest } from '../../../helpers/requests' 6import { doRequest } from '../../../helpers/requests'
6import { BROADCAST_CONCURRENCY } from '../../../initializers/constants' 7import { BROADCAST_CONCURRENCY } from '../../../initializers/constants'
7import { ActorFollowScoreCache } from '../../files-cache'
8import { buildGlobalHeaders, buildSignedRequestOptions, computeBody } from './utils/activitypub-http-utils' 8import { buildGlobalHeaders, buildSignedRequestOptions, computeBody } from './utils/activitypub-http-utils'
9 9
10async function processActivityPubHttpBroadcast (job: Job) { 10async function processActivityPubHttpBroadcast (job: Job) {
@@ -25,13 +25,17 @@ async function processActivityPubHttpBroadcast (job: Job) {
25 const badUrls: string[] = [] 25 const badUrls: string[] = []
26 const goodUrls: string[] = [] 26 const goodUrls: string[] = []
27 27
28 await map(payload.uris, uri => { 28 await map(payload.uris, async uri => {
29 return doRequest(uri, options) 29 try {
30 .then(() => goodUrls.push(uri)) 30 await doRequest(uri, options)
31 .catch(() => badUrls.push(uri)) 31 goodUrls.push(uri)
32 } catch (err) {
33 logger.debug('HTTP broadcast to %s failed.', uri, { err })
34 badUrls.push(uri)
35 }
32 }, { concurrency: BROADCAST_CONCURRENCY }) 36 }, { concurrency: BROADCAST_CONCURRENCY })
33 37
34 return ActorFollowScoreCache.Instance.updateActorFollowsScore(goodUrls, badUrls) 38 return ActorFollowHealthCache.Instance.updateActorFollowsHealth(goodUrls, badUrls)
35} 39}
36 40
37// --------------------------------------------------------------------------- 41// ---------------------------------------------------------------------------