Browse Source

Move server follow in the job queue

It helps to track follow errors
tags/v1.0.0-beta.4
Chocobozzz 4 years ago
parent
commit
5350fd8e5b
No known key found for this signature in database GPG Key ID: 583A612D890159BE
5 changed files with 90 additions and 64 deletions
  1. +15
    -63
      server/controllers/api/server/follows.ts
  2. +2
    -0
      server/initializers/constants.ts
  3. +68
    -0
      server/lib/job-queue/handlers/activitypub-follow.ts
  4. +4
    -1
      server/lib/job-queue/job-queue.ts
  5. +1
    -0
      shared/models/server/job.model.ts

+ 15
- 63
server/controllers/api/server/follows.ts View File

@@ -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<any>[] = []
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



+ 2
- 0
server/initializers/constants.ts View File

@@ -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
}


+ 68
- 0
server/lib/job-queue/handlers/activitypub-follow.ts View File

@@ -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)
})
}

+ 4
- 1
server/lib/job-queue/job-queue.ts View File

@@ -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<any>} = {
'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 })


+ 1
- 0
shared/models/server/job.model.ts View File

@@ -3,6 +3,7 @@ export type JobState = 'active' | 'complete' | 'failed' | 'inactive' | 'delayed'
export type JobType = 'activitypub-http-unicast' |
'activitypub-http-broadcast' |
'activitypub-http-fetcher' |
'activitypub-follow' |
'video-file' |
'email'



Loading…
Cancel
Save