job!
This commit is contained in:
		
							parent
							
								
									d2ea04fbf2
								
							
						
					
					
						commit
						239824c187
					
				|  | @ -133,16 +133,20 @@ id: 'aid' | |||
| #clusterLimit: 1 | ||||
| 
 | ||||
| # Job concurrency per worker | ||||
| # deliverJobConcurrency: 128 | ||||
| # inboxJobConcurrency: 16 | ||||
| #deliverJobConcurrency: 128 | ||||
| #inboxJobConcurrency: 16 | ||||
| #relashionshipJobConcurrency: 16 | ||||
| # What's relashionshipJob?: | ||||
| #  Follow, unfollow, block and unblock(ings) while following-imports, etc. or account migrations. | ||||
| 
 | ||||
| # Job rate limiter | ||||
| # deliverJobPerSec: 128 | ||||
| # inboxJobPerSec: 16 | ||||
| #deliverJobPerSec: 128 | ||||
| #inboxJobPerSec: 16 | ||||
| #relashionshipJobPerSec: 64 | ||||
| 
 | ||||
| # Job attempts | ||||
| # deliverJobMaxAttempts: 12 | ||||
| # inboxJobMaxAttempts: 8 | ||||
| #deliverJobMaxAttempts: 12 | ||||
| #inboxJobMaxAttempts: 8 | ||||
| 
 | ||||
| # IP address family used for outgoing request (ipv4, ipv6 or dual) | ||||
| #outgoingAddressFamily: ipv4 | ||||
|  |  | |||
|  | @ -6,5 +6,6 @@ | |||
|     "files.associations": { | ||||
|         "*.test.ts": "typescript" | ||||
|     }, | ||||
|     "jest.jestCommandLine": "pnpm run jest", | ||||
|     "jest.autoRun": "off" | ||||
| } | ||||
|  | @ -84,8 +84,10 @@ export type Source = { | |||
| 
 | ||||
| 	deliverJobConcurrency?: number; | ||||
| 	inboxJobConcurrency?: number; | ||||
| 	relashionshipJobConcurrency?: number; | ||||
| 	deliverJobPerSec?: number; | ||||
| 	inboxJobPerSec?: number; | ||||
| 	relashionshipJobPerSec?: number; | ||||
| 	deliverJobMaxAttempts?: number; | ||||
| 	inboxJobMaxAttempts?: number; | ||||
| 
 | ||||
|  |  | |||
|  | @ -96,14 +96,14 @@ export class AccountMoveService { | |||
| 		const iObj = await this.userEntityService.pack<true, true>(src.id, src, { detail: true, includeSecrets: true }); | ||||
| 		this.globalEventService.publishMainStream(src.id, 'meUpdated', iObj); | ||||
| 
 | ||||
| 		// Unfollow
 | ||||
| 		// Unfollow after 24 hours
 | ||||
| 		const followings = await this.followingsRepository.findBy({ | ||||
| 			followerId: src.id, | ||||
| 		}); | ||||
| 		this.queueService.createUnfollowJob(followings.map(following => ({ | ||||
| 		this.queueService.createDelayedUnfollowJob(followings.map(following => ({ | ||||
| 			from: { id: src.id }, | ||||
| 			to: { id: following.followeeId }, | ||||
| 		}))); | ||||
| 		})), process.env.NODE_ENV === 'test' ? 10000 : 1000 * 60 * 60 * 24); | ||||
| 
 | ||||
| 		await this.postMoveProcess(src, dst); | ||||
| 
 | ||||
|  |  | |||
|  | @ -78,7 +78,7 @@ const $db: Provider = { | |||
| 
 | ||||
| const $relationship: Provider = { | ||||
| 	provide: 'queue:relationship', | ||||
| 	useFactory: (config: Config) => q(config, 'relationship'), | ||||
| 	useFactory: (config: Config) => q(config, 'relationship', config.relashionshipJobPerSec ?? 64), | ||||
| 	inject: [DI.config], | ||||
| }; | ||||
| 
 | ||||
|  |  | |||
|  | @ -258,6 +258,12 @@ export class QueueService { | |||
| 		return this.relationshipQueue.addBulk(jobs); | ||||
| 	} | ||||
| 
 | ||||
| 	@bindThis | ||||
| 	public createDelayedUnfollowJob(followings: { from: ThinUser, to: ThinUser, requestId?: string }[], delay: number) { | ||||
| 		const jobs = followings.map(rel => this.generateRelationshipJobData('unfollow', rel, { delay })); | ||||
| 		return this.relationshipQueue.addBulk(jobs); | ||||
| 	} | ||||
| 
 | ||||
| 	@bindThis | ||||
| 	public createBlockJob(blockings: { from: ThinUser, to: ThinUser, silent?: boolean }[]) { | ||||
| 		const jobs = blockings.map(rel => this.generateRelationshipJobData('block', rel)); | ||||
|  | @ -271,7 +277,7 @@ export class QueueService { | |||
| 	} | ||||
| 
 | ||||
| 	@bindThis | ||||
| 	private generateRelationshipJobData(name: 'follow' | 'unfollow' | 'block' | 'unblock', data: RelationshipJobData): { | ||||
| 	private generateRelationshipJobData(name: 'follow' | 'unfollow' | 'block' | 'unblock', data: RelationshipJobData, opts?: Bull.JobOptions): { | ||||
| 		name: string, | ||||
| 		data: RelationshipJobData, | ||||
| 		opts: Bull.JobOptions, | ||||
|  | @ -287,6 +293,7 @@ export class QueueService { | |||
| 			opts: { | ||||
| 				removeOnComplete: true, | ||||
| 				removeOnFail: true, | ||||
| 				...opts, | ||||
| 			}, | ||||
| 		}; | ||||
| 	} | ||||
|  |  | |||
|  | @ -17,7 +17,7 @@ export class RelationshipQueueProcessorsService { | |||
| 
 | ||||
| 	@bindThis | ||||
| 	public start(q: Bull.Queue): void { | ||||
| 		const maxJobs = (this.config.deliverJobConcurrency ?? 128) / 4; // conservative?
 | ||||
| 		const maxJobs = this.config.relashionshipJobConcurrency ?? 16; | ||||
| 		q.process('follow', maxJobs, (job) => this.relationshipProcessorService.processFollow(job)); | ||||
| 		q.process('unfollow', maxJobs, (job) => this.relationshipProcessorService.processUnfollow(job)); | ||||
| 		q.process('block', maxJobs, (job) => this.relationshipProcessorService.processBlock(job)); | ||||
|  |  | |||
|  | @ -266,11 +266,12 @@ describe('Account Move', () => { | |||
| 
 | ||||
| 			await sleep(1000 * 3); // wait for jobs to finish
 | ||||
| 
 | ||||
| 			// Unfollow delayed?
 | ||||
| 			const aliceFollowings = await api('/users/following', { | ||||
| 				userId: alice.id, | ||||
| 			}, alice); | ||||
| 			assert.strictEqual(aliceFollowings.status, 200); | ||||
| 			assert.strictEqual(aliceFollowings.body.length, 0); | ||||
| 			assert.strictEqual(aliceFollowings.body.length, 3); | ||||
| 
 | ||||
| 			const carolFollowings = await api('/users/following', { | ||||
| 				userId: carol.id, | ||||
|  | @ -304,16 +305,35 @@ describe('Account Move', () => { | |||
| 			assert.ok(eveLists.body[0].userIds.find((id: string) => id === bob.id)); | ||||
| 		}); | ||||
| 
 | ||||
| 		test('Unable to move if the destination account has already moved.', async () => { | ||||
| 			await api('/i/move', { | ||||
| 				moveToAccount: `@bob@${url.hostname}`, | ||||
| 		test('A locked account automatically accept the follow request if it had already accepted the old account.', async () => { | ||||
| 			await successfulApiCall({ | ||||
| 				endpoint: '/following/create', | ||||
| 				parameters: { | ||||
| 					userId: frank.id, | ||||
| 				}, | ||||
| 				user: bob, | ||||
| 			}); | ||||
| 			const followers = await api('/users/followers', { | ||||
| 				userId: frank.id, | ||||
| 			}, frank); | ||||
| 
 | ||||
| 			assert.strictEqual(followers.status, 200); | ||||
| 			assert.strictEqual(followers.body.length, 2); | ||||
| 			assert.strictEqual(followers.body[0].followerId, bob.id); | ||||
| 		}); | ||||
| 
 | ||||
| 		test('Unfollowed after 10 sec (24 hours in production).', async () => { | ||||
| 			await sleep(1000 * 8); | ||||
| 
 | ||||
| 			const following = await api('/users/following', { | ||||
| 				userId: alice.id, | ||||
| 			}, alice); | ||||
| 
 | ||||
| 			const newAlice = await Users.findOneByOrFail({ id: alice.id }); | ||||
| 			assert.strictEqual(newAlice.movedToUri, `${url.origin}/users/${bob.id}`); | ||||
| 			assert.strictEqual(newAlice.alsoKnownAs?.length, 1); | ||||
| 			assert.strictEqual(newAlice.alsoKnownAs[0], `${url.origin}/users/${bob.id}`); | ||||
| 			assert.strictEqual(following.status, 200); | ||||
| 			assert.strictEqual(following.body.length, 0); | ||||
| 		}); | ||||
| 
 | ||||
| 		test('Unable to move if the destination account has already moved.', async () => { | ||||
| 			const res = await api('/i/move', { | ||||
| 				moveToAccount: `@alice@${url.hostname}`, | ||||
| 			}, bob); | ||||
|  | @ -345,23 +365,6 @@ describe('Account Move', () => { | |||
| 			assert.strictEqual(newEve.followersCount, 1); | ||||
| 		}); | ||||
| 
 | ||||
| 		test('A locked account automatically accept the follow request if it had already accepted the old account.', async () => { | ||||
| 			await successfulApiCall({ | ||||
| 				endpoint: '/following/create', | ||||
| 				parameters: { | ||||
| 					userId: frank.id, | ||||
| 				}, | ||||
| 				user: bob, | ||||
| 			}); | ||||
| 			const followers = await api('/users/followers', { | ||||
| 				userId: frank.id, | ||||
| 			}, frank); | ||||
| 
 | ||||
| 			assert.strictEqual(followers.status, 200); | ||||
| 			assert.strictEqual(followers.body.length, 2); | ||||
| 			assert.strictEqual(followers.body[0].followerId, bob.id); | ||||
| 		}); | ||||
| 
 | ||||
| 		test.each([ | ||||
| 			'/antennas/create', | ||||
| 			'/channels/create', | ||||
|  |  | |||
		Loading…
	
		Reference in New Issue