[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

1 files changed, 52 insertions, 52 deletions

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: &gtsmodel.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)