From c9c0773f2c2363dcfa37e675b83ec3f0b49bd0d9 Mon Sep 17 00:00:00 2001 From: kim <89579420+NyaaaWhatsUpDoc@users.noreply.github.com> Date: Fri, 26 Apr 2024 13:50:46 +0100 Subject: [performance] update remaining worker pools to use queues (#2865) * start replacing client + federator + media workers with new worker + queue types * refactor federatingDB.Delete(), drop queued messages when deleting account / status * move all queue purging to the processor workers * undo toolchain updates * code comments, ensure dereferencer worker pool gets started * update gruf libraries in readme * start the job scheduler separately to the worker pools * reshuffle ordering or server.go + remove duplicate worker start / stop * update go-list version * fix vendoring * move queue invalidation to before wipeing / deletion, to ensure queued work not dropped * add logging to worker processing functions in testrig, don't start workers in unexpected places * update go-structr to add (+then rely on) QueueCtx{} type * ensure more worker pools get started properly in tests * fix remaining broken tests relying on worker queue logic * fix account test suite queue popping logic, ensure noop workers do not pull from queue * move back accidentally shuffled account deletion order * ensure error (non nil!!) gets passed in refactored federatingDB{}.Delete() * silently drop deletes from accounts not permitted to * don't warn log on forwarded deletes * make if else clauses easier to parse * use getFederatorMsg() * improved code comment * improved code comment re: requesting account delete checks * remove boolean result from worker start / stop since false = already running or already stopped * remove optional passed-in http.client * remove worker starting from the admin CLI commands (we don't need to handle side-effects) * update prune cli to start scheduler but not all of the workers * fix rebase issues * remove redundant return statements * i'm sorry sir linter --- internal/processing/workers/fromfediapi_test.go | 104 ++++++++++++------------ 1 file changed, 52 insertions(+), 52 deletions(-) (limited to 'internal/processing/workers/fromfediapi_test.go') diff --git a/internal/processing/workers/fromfediapi_test.go b/internal/processing/workers/fromfediapi_test.go index eb3d73e0c..b8acca94d 100644 --- a/internal/processing/workers/fromfediapi_test.go +++ b/internal/processing/workers/fromfediapi_test.go @@ -54,12 +54,12 @@ func (suite *FromFediAPITestSuite) TestProcessFederationAnnounce() { announceStatus.Account = boostingAccount announceStatus.Visibility = boostedStatus.Visibility - err := suite.processor.Workers().ProcessFromFediAPI(context.Background(), messages.FromFediAPI{ - APObjectType: ap.ActivityAnnounce, - APActivityType: ap.ActivityCreate, - GTSModel: announceStatus, - ReceivingAccount: suite.testAccounts["local_account_1"], - RequestingAccount: boostingAccount, + err := suite.processor.Workers().ProcessFromFediAPI(context.Background(), &messages.FromFediAPI{ + APObjectType: ap.ActivityAnnounce, + APActivityType: ap.ActivityCreate, + GTSModel: announceStatus, + Receiving: suite.testAccounts["local_account_1"], + Requesting: boostingAccount, }) suite.NoError(err) @@ -115,12 +115,12 @@ func (suite *FromFediAPITestSuite) TestProcessReplyMention() { suite.NoError(errWithCode) // Send the replied status off to the fedi worker to be further processed. - err = suite.processor.Workers().ProcessFromFediAPI(context.Background(), messages.FromFediAPI{ - APObjectType: ap.ObjectNote, - APActivityType: ap.ActivityCreate, - APObjectModel: replyingStatusable, - ReceivingAccount: repliedAccount, - RequestingAccount: replyingAccount, + err = suite.processor.Workers().ProcessFromFediAPI(context.Background(), &messages.FromFediAPI{ + APObjectType: ap.ObjectNote, + APActivityType: ap.ActivityCreate, + APObject: replyingStatusable, + Receiving: repliedAccount, + Requesting: replyingAccount, }) suite.NoError(err) @@ -179,12 +179,12 @@ func (suite *FromFediAPITestSuite) TestProcessFave() { err := suite.db.Put(context.Background(), fave) suite.NoError(err) - err = suite.processor.Workers().ProcessFromFediAPI(context.Background(), messages.FromFediAPI{ - APObjectType: ap.ActivityLike, - APActivityType: ap.ActivityCreate, - GTSModel: fave, - ReceivingAccount: favedAccount, - RequestingAccount: favingAccount, + err = suite.processor.Workers().ProcessFromFediAPI(context.Background(), &messages.FromFediAPI{ + APObjectType: ap.ActivityLike, + APActivityType: ap.ActivityCreate, + GTSModel: fave, + Receiving: favedAccount, + Requesting: favingAccount, }) suite.NoError(err) @@ -249,12 +249,12 @@ func (suite *FromFediAPITestSuite) TestProcessFaveWithDifferentReceivingAccount( err := suite.db.Put(context.Background(), fave) suite.NoError(err) - err = suite.processor.Workers().ProcessFromFediAPI(context.Background(), messages.FromFediAPI{ - APObjectType: ap.ActivityLike, - APActivityType: ap.ActivityCreate, - GTSModel: fave, - ReceivingAccount: receivingAccount, - RequestingAccount: favingAccount, + err = suite.processor.Workers().ProcessFromFediAPI(context.Background(), &messages.FromFediAPI{ + APObjectType: ap.ActivityLike, + APActivityType: ap.ActivityCreate, + GTSModel: fave, + Receiving: receivingAccount, + Requesting: favingAccount, }) suite.NoError(err) @@ -321,12 +321,12 @@ func (suite *FromFediAPITestSuite) TestProcessAccountDelete() { suite.NoError(err) // now they are mufos! - err = suite.processor.Workers().ProcessFromFediAPI(ctx, messages.FromFediAPI{ - APObjectType: ap.ObjectProfile, - APActivityType: ap.ActivityDelete, - GTSModel: deletedAccount, - ReceivingAccount: receivingAccount, - RequestingAccount: deletedAccount, + err = suite.processor.Workers().ProcessFromFediAPI(ctx, &messages.FromFediAPI{ + APObjectType: ap.ObjectProfile, + APActivityType: ap.ActivityDelete, + GTSModel: deletedAccount, + Receiving: receivingAccount, + Requesting: deletedAccount, }) suite.NoError(err) @@ -402,12 +402,12 @@ func (suite *FromFediAPITestSuite) TestProcessFollowRequestLocked() { err := suite.db.Put(ctx, satanFollowRequestTurtle) suite.NoError(err) - err = suite.processor.Workers().ProcessFromFediAPI(ctx, messages.FromFediAPI{ - APObjectType: ap.ActivityFollow, - APActivityType: ap.ActivityCreate, - GTSModel: satanFollowRequestTurtle, - ReceivingAccount: targetAccount, - RequestingAccount: originAccount, + err = suite.processor.Workers().ProcessFromFediAPI(ctx, &messages.FromFediAPI{ + APObjectType: ap.ActivityFollow, + APActivityType: ap.ActivityCreate, + GTSModel: satanFollowRequestTurtle, + Receiving: targetAccount, + Requesting: originAccount, }) suite.NoError(err) @@ -456,12 +456,12 @@ func (suite *FromFediAPITestSuite) TestProcessFollowRequestUnlocked() { err := suite.db.Put(ctx, satanFollowRequestTurtle) suite.NoError(err) - err = suite.processor.Workers().ProcessFromFediAPI(ctx, messages.FromFediAPI{ - APObjectType: ap.ActivityFollow, - APActivityType: ap.ActivityCreate, - GTSModel: satanFollowRequestTurtle, - ReceivingAccount: targetAccount, - RequestingAccount: originAccount, + err = suite.processor.Workers().ProcessFromFediAPI(ctx, &messages.FromFediAPI{ + APObjectType: ap.ActivityFollow, + APActivityType: ap.ActivityCreate, + GTSModel: satanFollowRequestTurtle, + Receiving: targetAccount, + Requesting: originAccount, }) suite.NoError(err) @@ -532,13 +532,13 @@ func (suite *FromFediAPITestSuite) TestCreateStatusFromIRI() { receivingAccount := suite.testAccounts["local_account_1"] statusCreator := suite.testAccounts["remote_account_2"] - err := suite.processor.Workers().ProcessFromFediAPI(ctx, messages.FromFediAPI{ - APObjectType: ap.ObjectNote, - APActivityType: ap.ActivityCreate, - GTSModel: nil, // gtsmodel is nil because this is a forwarded status -- we want to dereference it using the iri - ReceivingAccount: receivingAccount, - RequestingAccount: statusCreator, - APIri: testrig.URLMustParse("http://example.org/users/Some_User/statuses/afaba698-5740-4e32-a702-af61aa543bc1"), + err := suite.processor.Workers().ProcessFromFediAPI(ctx, &messages.FromFediAPI{ + APObjectType: ap.ObjectNote, + APActivityType: ap.ActivityCreate, + GTSModel: nil, // gtsmodel is nil because this is a forwarded status -- we want to dereference it using the iri + Receiving: receivingAccount, + Requesting: statusCreator, + APIRI: testrig.URLMustParse("http://example.org/users/Some_User/statuses/afaba698-5740-4e32-a702-af61aa543bc1"), }) suite.NoError(err) @@ -585,7 +585,7 @@ func (suite *FromFediAPITestSuite) TestMoveAccount() { } // Process the Move. - err := suite.processor.Workers().ProcessFromFediAPI(ctx, messages.FromFediAPI{ + err := suite.processor.Workers().ProcessFromFediAPI(ctx, &messages.FromFediAPI{ APObjectType: ap.ObjectProfile, APActivityType: ap.ActivityMove, GTSModel: >smodel.Move{ @@ -595,8 +595,8 @@ func (suite *FromFediAPITestSuite) TestMoveAccount() { Target: testrig.URLMustParse(targetAcct.URI), URI: "https://fossbros-anonymous.io/users/foss_satan/moves/01HRA064871MR8HGVSAFJ333GM", }, - ReceivingAccount: receivingAcct, - RequestingAccount: requestingAcct, + Receiving: receivingAcct, + Requesting: requestingAcct, }) suite.NoError(err) -- cgit v1.2.3