[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, 12 insertions, 12 deletions

diff --git a/internal/processing/account/follow.go b/internal/processing/account/follow.go
index 7f28bb2c0..402c764e8 100644
--- a/internal/processing/account/follow.go
+++ b/internal/processing/account/follow.go
@@ -117,12 +117,12 @@ func (p *Processor) FollowCreate(ctx context.Context, requestingAccount *gtsmode
 	} else {
 		// Otherwise we leave the follow request as it is,
 		// and we handle the rest of the process async.
-		p.state.Workers.EnqueueClientAPI(ctx, messages.FromClientAPI{
+		p.state.Workers.Client.Queue.Push(&messages.FromClientAPI{
 			APObjectType:   ap.ActivityFollow,
 			APActivityType: ap.ActivityCreate,
 			GTSModel:       fr,
-			OriginAccount:  requestingAccount,
-			TargetAccount:  targetAccount,
+			Origin:         requestingAccount,
+			Target:         targetAccount,
 		})
 	}
 
@@ -143,7 +143,7 @@ func (p *Processor) FollowRemove(ctx context.Context, requestingAccount *gtsmode
 	}
 
 	// Batch queue accreted client api messages.
-	p.state.Workers.EnqueueClientAPI(ctx, msgs...)
+	p.state.Workers.Client.Queue.Push(msgs...)
 
 	return p.RelationshipGet(ctx, requestingAccount, targetAccountID)
 }
@@ -225,8 +225,8 @@ func (p *Processor) getFollowTarget(ctx context.Context, requester *gtsmodel.Acc
 // If a follow and/or follow request was removed this way, one or two
 // messages will be returned which should then be processed by a client
 // api worker.
-func (p *Processor) unfollow(ctx context.Context, requestingAccount *gtsmodel.Account, targetAccount *gtsmodel.Account) ([]messages.FromClientAPI, error) {
-	var msgs []messages.FromClientAPI
+func (p *Processor) unfollow(ctx context.Context, requestingAccount *gtsmodel.Account, targetAccount *gtsmodel.Account) ([]*messages.FromClientAPI, error) {
+	var msgs []*messages.FromClientAPI
 
 	// Get follow from requesting account to target account.
 	follow, err := p.state.DB.GetFollow(ctx, requestingAccount.ID, targetAccount.ID)
@@ -251,7 +251,7 @@ func (p *Processor) unfollow(ctx context.Context, requestingAccount *gtsmodel.Ac
 		}
 
 		// Follow status changed, process side effects.
-		msgs = append(msgs, messages.FromClientAPI{
+		msgs = append(msgs, &messages.FromClientAPI{
 			APObjectType:   ap.ActivityFollow,
 			APActivityType: ap.ActivityUndo,
 			GTSModel: &gtsmodel.Follow{
@@ -259,8 +259,8 @@ func (p *Processor) unfollow(ctx context.Context, requestingAccount *gtsmodel.Ac
 				TargetAccountID: targetAccount.ID,
 				URI:             follow.URI,
 			},
-			OriginAccount: requestingAccount,
-			TargetAccount: targetAccount,
+			Origin: requestingAccount,
+			Target: targetAccount,
 		})
 	}
 
@@ -287,7 +287,7 @@ func (p *Processor) unfollow(ctx context.Context, requestingAccount *gtsmodel.Ac
 		}
 
 		// Follow status changed, process side effects.
-		msgs = append(msgs, messages.FromClientAPI{
+		msgs = append(msgs, &messages.FromClientAPI{
 			APObjectType:   ap.ActivityFollow,
 			APActivityType: ap.ActivityUndo,
 			GTSModel: &gtsmodel.Follow{
@@ -295,8 +295,8 @@ func (p *Processor) unfollow(ctx context.Context, requestingAccount *gtsmodel.Ac
 				TargetAccountID: targetAccount.ID,
 				URI:             followReq.URI,
 			},
-			OriginAccount: requestingAccount,
-			TargetAccount: targetAccount,
+			Origin: requestingAccount,
+			Target: targetAccount,
 		})
 	}