From 5350fd8e5b2b2d017b16d97828893a8a4a40bd89 Mon Sep 17 00:00:00 2001 From: Chocobozzz Date: Wed, 18 Apr 2018 15:32:40 +0200 Subject: Move server follow in the job queue It helps to track follow errors --- server/controllers/api/server/follows.ts | 78 +++++----------------- server/initializers/constants.ts | 2 + .../lib/job-queue/handlers/activitypub-follow.ts | 68 +++++++++++++++++++ server/lib/job-queue/job-queue.ts | 5 +- 4 files changed, 89 insertions(+), 64 deletions(-) create mode 100644 server/lib/job-queue/handlers/activitypub-follow.ts (limited to 'server') diff --git a/server/controllers/api/server/follows.ts b/server/controllers/api/server/follows.ts index bb0063473..e78361c9a 100644 --- a/server/controllers/api/server/follows.ts +++ b/server/controllers/api/server/follows.ts @@ -1,20 +1,22 @@ import * as express from 'express' import { UserRight } from '../../../../shared/models/users' -import { sanitizeHost } from '../../../helpers/core-utils' -import { retryTransactionWrapper } from '../../../helpers/database-utils' import { logger } from '../../../helpers/logger' import { getFormattedObjects, getServerActor } from '../../../helpers/utils' -import { loadActorUrlOrGetFromWebfinger } from '../../../helpers/webfinger' -import { REMOTE_SCHEME, sequelizeTypescript, SERVER_ACTOR_NAME } from '../../../initializers' -import { getOrCreateActorAndServerAndModel } from '../../../lib/activitypub/actor' -import { sendFollow, sendUndoFollow } from '../../../lib/activitypub/send' +import { sequelizeTypescript } from '../../../initializers' +import { sendUndoFollow } from '../../../lib/activitypub/send' import { - asyncMiddleware, authenticate, ensureUserHasRight, paginationValidator, removeFollowingValidator, setBodyHostsPort, setDefaultSort, - setDefaultPagination + asyncMiddleware, + authenticate, + ensureUserHasRight, + paginationValidator, + removeFollowingValidator, + setBodyHostsPort, + setDefaultPagination, + setDefaultSort } from '../../../middlewares' import { followersSortValidator, followingSortValidator, followValidator } from '../../../middlewares/validators' -import { ActorModel } from '../../../models/activitypub/actor' import { ActorFollowModel } from '../../../models/activitypub/actor-follow' +import { JobQueue } from '../../../lib/job-queue' const serverFollowsRouter = express.Router() serverFollowsRouter.get('/following', @@ -30,7 +32,7 @@ serverFollowsRouter.post('/following', ensureUserHasRight(UserRight.MANAGE_SERVER_FOLLOW), followValidator, setBodyHostsPort, - asyncMiddleware(followRetry) + asyncMiddleware(followInstance) ) serverFollowsRouter.delete('/following/:host', @@ -70,67 +72,17 @@ async function listFollowers (req: express.Request, res: express.Response, next: return res.json(getFormattedObjects(resultList.data, resultList.total)) } -async function followRetry (req: express.Request, res: express.Response, next: express.NextFunction) { +async function followInstance (req: express.Request, res: express.Response, next: express.NextFunction) { const hosts = req.body.hosts as string[] - const fromActor = await getServerActor() - - const tasks: Promise[] = [] - const actorName = SERVER_ACTOR_NAME for (const host of hosts) { - const sanitizedHost = sanitizeHost(host, REMOTE_SCHEME.HTTP) - - // We process each host in a specific transaction - // First, we add the follow request in the database - // Then we send the follow request to other actor - const p = loadActorUrlOrGetFromWebfinger(actorName, sanitizedHost) - .then(actorUrl => getOrCreateActorAndServerAndModel(actorUrl)) - .then(targetActor => { - const options = { - arguments: [ fromActor, targetActor ], - errorMessage: 'Cannot follow with many retries.' - } - - return retryTransactionWrapper(follow, options) - }) - .catch(err => logger.warn('Cannot follow server %s.', sanitizedHost, { err })) - - tasks.push(p) + JobQueue.Instance.createJob({ type: 'activitypub-follow', payload: { host } }) + .catch(err => logger.error('Cannot create follow job for %s.', host, err)) } - // Don't make the client wait the tasks - Promise.all(tasks) - .catch(err => logger.error('Error in follow.', { err })) - return res.status(204).end() } -function follow (fromActor: ActorModel, targetActor: ActorModel) { - if (fromActor.id === targetActor.id) { - throw new Error('Follower is the same than target actor.') - } - - return sequelizeTypescript.transaction(async t => { - const [ actorFollow ] = await ActorFollowModel.findOrCreate({ - where: { - actorId: fromActor.id, - targetActorId: targetActor.id - }, - defaults: { - state: 'pending', - actorId: fromActor.id, - targetActorId: targetActor.id - }, - transaction: t - }) - actorFollow.ActorFollowing = targetActor - actorFollow.ActorFollower = fromActor - - // Send a notification to remote server - await sendFollow(actorFollow) - }) -} - async function removeFollow (req: express.Request, res: express.Response, next: express.NextFunction) { const follow: ActorFollowModel = res.locals.follow diff --git a/server/initializers/constants.ts b/server/initializers/constants.ts index 9fde989c5..ffcbe69b8 100644 --- a/server/initializers/constants.ts +++ b/server/initializers/constants.ts @@ -65,6 +65,7 @@ const JOB_ATTEMPTS: { [ id in JobType ]: number } = { 'activitypub-http-broadcast': 5, 'activitypub-http-unicast': 5, 'activitypub-http-fetcher': 5, + 'activitypub-follow': 5, 'video-file': 1, 'email': 5 } @@ -72,6 +73,7 @@ const JOB_CONCURRENCY: { [ id in JobType ]: number } = { 'activitypub-http-broadcast': 1, 'activitypub-http-unicast': 5, 'activitypub-http-fetcher': 1, + 'activitypub-follow': 3, 'video-file': 1, 'email': 5 } diff --git a/server/lib/job-queue/handlers/activitypub-follow.ts b/server/lib/job-queue/handlers/activitypub-follow.ts new file mode 100644 index 000000000..6764a4037 --- /dev/null +++ b/server/lib/job-queue/handlers/activitypub-follow.ts @@ -0,0 +1,68 @@ +import * as kue from 'kue' +import { logger } from '../../../helpers/logger' +import { getServerActor } from '../../../helpers/utils' +import { REMOTE_SCHEME, sequelizeTypescript, SERVER_ACTOR_NAME } from '../../../initializers' +import { sendFollow } from '../../activitypub/send' +import { sanitizeHost } from '../../../helpers/core-utils' +import { loadActorUrlOrGetFromWebfinger } from '../../../helpers/webfinger' +import { getOrCreateActorAndServerAndModel } from '../../activitypub/actor' +import { retryTransactionWrapper } from '../../../helpers/database-utils' +import { ActorFollowModel } from '../../../models/activitypub/actor-follow' +import { ActorModel } from '../../../models/activitypub/actor' + +export type ActivitypubFollowPayload = { + host: string +} + +async function processActivityPubFollow (job: kue.Job) { + const payload = job.data as ActivitypubFollowPayload + const host = payload.host + + logger.info('Processing ActivityPub follow in job %d.', job.id) + + const sanitizedHost = sanitizeHost(host, REMOTE_SCHEME.HTTP) + + const actorUrl = await loadActorUrlOrGetFromWebfinger(SERVER_ACTOR_NAME, sanitizedHost) + const targetActor = await getOrCreateActorAndServerAndModel(actorUrl) + + const fromActor = await getServerActor() + const options = { + arguments: [ fromActor, targetActor ], + errorMessage: 'Cannot follow with many retries.' + } + + return retryTransactionWrapper(follow, options) +} +// --------------------------------------------------------------------------- + +export { + processActivityPubFollow +} + +// --------------------------------------------------------------------------- + +function follow (fromActor: ActorModel, targetActor: ActorModel) { + if (fromActor.id === targetActor.id) { + throw new Error('Follower is the same than target actor.') + } + + return sequelizeTypescript.transaction(async t => { + const [ actorFollow ] = await ActorFollowModel.findOrCreate({ + where: { + actorId: fromActor.id, + targetActorId: targetActor.id + }, + defaults: { + state: 'pending', + actorId: fromActor.id, + targetActorId: targetActor.id + }, + transaction: t + }) + actorFollow.ActorFollowing = targetActor + actorFollow.ActorFollower = fromActor + + // Send a notification to remote server + await sendFollow(actorFollow) + }) +} diff --git a/server/lib/job-queue/job-queue.ts b/server/lib/job-queue/job-queue.ts index 1dc28755e..bf40a9206 100644 --- a/server/lib/job-queue/job-queue.ts +++ b/server/lib/job-queue/job-queue.ts @@ -8,11 +8,13 @@ import { ActivitypubHttpFetcherPayload, processActivityPubHttpFetcher } from './ import { ActivitypubHttpUnicastPayload, processActivityPubHttpUnicast } from './handlers/activitypub-http-unicast' import { EmailPayload, processEmail } from './handlers/email' import { processVideoFile, VideoFilePayload } from './handlers/video-file' +import { ActivitypubFollowPayload, processActivityPubFollow } from './handlers/activitypub-follow' type CreateJobArgument = { type: 'activitypub-http-broadcast', payload: ActivitypubHttpBroadcastPayload } | { type: 'activitypub-http-unicast', payload: ActivitypubHttpUnicastPayload } | { type: 'activitypub-http-fetcher', payload: ActivitypubHttpFetcherPayload } | + { type: 'activitypub-follow', payload: ActivitypubFollowPayload } | { type: 'video-file', payload: VideoFilePayload } | { type: 'email', payload: EmailPayload } @@ -20,6 +22,7 @@ const handlers: { [ id in JobType ]: (job: kue.Job) => Promise} = { 'activitypub-http-broadcast': processActivityPubHttpBroadcast, 'activitypub-http-unicast': processActivityPubHttpUnicast, 'activitypub-http-fetcher': processActivityPubHttpFetcher, + 'activitypub-follow': processActivityPubFollow, 'video-file': processVideoFile, 'email': processEmail } @@ -50,7 +53,7 @@ class JobQueue { } }) - this.jobQueue.setMaxListeners(15) + this.jobQueue.setMaxListeners(20) this.jobQueue.on('error', err => { logger.error('Error in job queue.', { err }) -- cgit v1.2.3