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

diff --git a/internal/federation/federatingdb/create.go b/internal/federation/federatingdb/create.go
index 94261526e..44f3cd98c 100644
--- a/internal/federation/federatingdb/create.go
+++ b/internal/federation/federatingdb/create.go
@@ -99,7 +99,7 @@ func (f *federatingDB) Create(ctx context.Context, asType vocab.Type) error {
 	BLOCK HANDLERS
 */
 
-func (f *federatingDB) activityBlock(ctx context.Context, asType vocab.Type, receiving *gtsmodel.Account, requestingAccount *gtsmodel.Account) error {
+func (f *federatingDB) activityBlock(ctx context.Context, asType vocab.Type, receiving *gtsmodel.Account, requesting *gtsmodel.Account) error {
 	blockable, ok := asType.(vocab.ActivityStreamsBlock)
 	if !ok {
 		return errors.New("activityBlock: could not convert type to block")
@@ -110,10 +110,10 @@ func (f *federatingDB) activityBlock(ctx context.Context, asType vocab.Type, rec
 		return fmt.Errorf("activityBlock: could not convert Block to gts model block")
 	}
 
-	if block.AccountID != requestingAccount.ID {
+	if block.AccountID != requesting.ID {
 		return fmt.Errorf(
 			"activityBlock: requestingAccount %s is not Block actor account %s",
-			requestingAccount.URI, block.Account.URI,
+			requesting.URI, block.Account.URI,
 		)
 	}
 
@@ -130,12 +130,12 @@ func (f *federatingDB) activityBlock(ctx context.Context, asType vocab.Type, rec
 		return fmt.Errorf("activityBlock: database error inserting block: %s", err)
 	}
 
-	f.state.Workers.EnqueueFediAPI(ctx, messages.FromFediAPI{
-		APObjectType:      ap.ActivityBlock,
-		APActivityType:    ap.ActivityCreate,
-		GTSModel:          block,
-		ReceivingAccount:  receiving,
-		RequestingAccount: requestingAccount,
+	f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{
+		APObjectType:   ap.ActivityBlock,
+		APActivityType: ap.ActivityCreate,
+		GTSModel:       block,
+		Receiving:      receiving,
+		Requesting:     requesting,
 	})
 
 	return nil
@@ -297,7 +297,7 @@ func (f *federatingDB) createPollOptionables(
 	}
 
 	// Enqueue message to the fedi API worker with poll vote(s).
-	f.state.Workers.EnqueueFediAPI(ctx, messages.FromFediAPI{
+	f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{
 		APActivityType: ap.ActivityCreate,
 		APObjectType:   ap.ActivityQuestion,
 		GTSModel: &gtsmodel.PollVote{
@@ -308,8 +308,8 @@ func (f *federatingDB) createPollOptionables(
 			PollID:    inReplyTo.PollID,
 			Poll:      inReplyTo.Poll,
 		},
-		ReceivingAccount:  receiver,
-		RequestingAccount: requester,
+		Receiving:  receiver,
+		Requesting: requester,
 	})
 
 	return nil
@@ -377,28 +377,28 @@ func (f *federatingDB) createStatusable(
 
 		// Pass the statusable URI (APIri) into the processor
 		// worker and do the rest of the processing asynchronously.
-		f.state.Workers.EnqueueFediAPI(ctx, messages.FromFediAPI{
-			APObjectType:      ap.ObjectNote,
-			APActivityType:    ap.ActivityCreate,
-			APIri:             ap.GetJSONLDId(statusable),
-			APObjectModel:     nil,
-			GTSModel:          nil,
-			ReceivingAccount:  receiver,
-			RequestingAccount: requester,
+		f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{
+			APObjectType:   ap.ObjectNote,
+			APActivityType: ap.ActivityCreate,
+			APIRI:          ap.GetJSONLDId(statusable),
+			APObject:       nil,
+			GTSModel:       nil,
+			Receiving:      receiver,
+			Requesting:     requester,
 		})
 		return nil
 	}
 
 	// Do the rest of the processing asynchronously. The processor
 	// will handle inserting/updating + further dereferencing the status.
-	f.state.Workers.EnqueueFediAPI(ctx, messages.FromFediAPI{
-		APObjectType:      ap.ObjectNote,
-		APActivityType:    ap.ActivityCreate,
-		APIri:             nil,
-		GTSModel:          nil,
-		APObjectModel:     statusable,
-		ReceivingAccount:  receiver,
-		RequestingAccount: requester,
+	f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{
+		APObjectType:   ap.ObjectNote,
+		APActivityType: ap.ActivityCreate,
+		APIRI:          nil,
+		GTSModel:       nil,
+		APObject:       statusable,
+		Receiving:      receiver,
+		Requesting:     requester,
 	})
 
 	return nil
@@ -439,12 +439,12 @@ func (f *federatingDB) activityFollow(ctx context.Context, asType vocab.Type, re
 		return fmt.Errorf("activityFollow: database error inserting follow request: %s", err)
 	}
 
-	f.state.Workers.EnqueueFediAPI(ctx, messages.FromFediAPI{
-		APObjectType:      ap.ActivityFollow,
-		APActivityType:    ap.ActivityCreate,
-		GTSModel:          followRequest,
-		ReceivingAccount:  receivingAccount,
-		RequestingAccount: requestingAccount,
+	f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{
+		APObjectType:   ap.ActivityFollow,
+		APActivityType: ap.ActivityCreate,
+		GTSModel:       followRequest,
+		Receiving:      receivingAccount,
+		Requesting:     requestingAccount,
 	})
 
 	return nil
@@ -484,12 +484,12 @@ func (f *federatingDB) activityLike(ctx context.Context, asType vocab.Type, rece
 		return fmt.Errorf("activityLike: database error inserting fave: %w", err)
 	}
 
-	f.state.Workers.EnqueueFediAPI(ctx, messages.FromFediAPI{
-		APObjectType:      ap.ActivityLike,
-		APActivityType:    ap.ActivityCreate,
-		GTSModel:          fave,
-		ReceivingAccount:  receivingAccount,
-		RequestingAccount: requestingAccount,
+	f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{
+		APObjectType:   ap.ActivityLike,
+		APActivityType: ap.ActivityCreate,
+		GTSModel:       fave,
+		Receiving:      receivingAccount,
+		Requesting:     requestingAccount,
 	})
 
 	return nil
@@ -536,12 +536,12 @@ func (f *federatingDB) activityFlag(ctx context.Context, asType vocab.Type, rece
 		return fmt.Errorf("activityFlag: database error inserting report: %w", err)
 	}
 
-	f.state.Workers.EnqueueFediAPI(ctx, messages.FromFediAPI{
-		APObjectType:      ap.ActivityFlag,
-		APActivityType:    ap.ActivityCreate,
-		GTSModel:          report,
-		ReceivingAccount:  receivingAccount,
-		RequestingAccount: requestingAccount,
+	f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{
+		APObjectType:   ap.ActivityFlag,
+		APActivityType: ap.ActivityCreate,
+		GTSModel:       report,
+		Receiving:      receivingAccount,
+		Requesting:     requestingAccount,
 	})
 
 	return nil