[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

9 files changed, 266 insertions, 244 deletions

diff --git a/internal/processing/workers/federate.go b/internal/processing/workers/federate.go
index e737513f5..3538c9958 100644
--- a/internal/processing/workers/federate.go
+++ b/internal/processing/workers/federate.go
@@ -75,12 +75,6 @@ func (f *federate) DeleteAccount(ctx context.Context, account *gtsmodel.Account)
 		return nil
 	}
 
-	// Drop any queued outgoing AP requests to / from account,
-	// (this stops any queued likes, boosts, creates etc).
-	f.state.Workers.Delivery.Queue.Delete("ActorID", account.URI)
-	f.state.Workers.Delivery.Queue.Delete("ObjectID", account.URI)
-	f.state.Workers.Delivery.Queue.Delete("TargetID", account.URI)
-
 	// Parse relevant URI(s).
 	outboxIRI, err := parseURI(account.OutboxURI)
 	if err != nil {
@@ -228,16 +222,6 @@ func (f *federate) DeleteStatus(ctx context.Context, status *gtsmodel.Status) er
 		return nil
 	}
 
-	// Drop any queued outgoing http requests for status,
-	// (this stops any queued likes, boosts, creates etc).
-	f.state.Workers.Delivery.Queue.Delete("ObjectID", status.URI)
-	f.state.Workers.Delivery.Queue.Delete("TargetID", status.URI)
-
-	// Ensure the status model is fully populated.
-	if err := f.state.DB.PopulateStatus(ctx, status); err != nil {
-		return gtserror.Newf("error populating status: %w", err)
-	}
-
 	// Parse the outbox URI of the status author.
 	outboxIRI, err := parseURI(status.Account.OutboxURI)
 	if err != nil {
diff --git a/internal/processing/workers/fromclientapi.go b/internal/processing/workers/fromclientapi.go
index 1412ea003..4564afbb9 100644
--- a/internal/processing/workers/fromclientapi.go
+++ b/internal/processing/workers/fromclientapi.go
@@ -25,7 +25,6 @@ import (
 	"codeberg.org/gruf/go-logger/v2/level"
 	"github.com/superseriousbusiness/gotosocial/internal/ap"
 	"github.com/superseriousbusiness/gotosocial/internal/db"
-	"github.com/superseriousbusiness/gotosocial/internal/gtscontext"
 	"github.com/superseriousbusiness/gotosocial/internal/gtserror"
 	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
 	"github.com/superseriousbusiness/gotosocial/internal/log"
@@ -45,29 +44,15 @@ type clientAPI struct {
 	surface   *surface
 	federate  *federate
 	account   *account.Processor
-	utilF     *utilF
+	utils     *utils
 }
 
-func (p *Processor) EnqueueClientAPI(cctx context.Context, msgs ...messages.FromClientAPI) {
-	_ = p.workers.ClientAPI.MustEnqueueCtx(cctx, func(wctx context.Context) {
-		// Copy caller ctx values to worker's.
-		wctx = gtscontext.WithValues(wctx, cctx)
-
-		// Process worker messages.
-		for _, msg := range msgs {
-			if err := p.ProcessFromClientAPI(wctx, msg); err != nil {
-				log.Errorf(wctx, "error processing client API message: %v", err)
-			}
-		}
-	})
-}
-
-func (p *Processor) ProcessFromClientAPI(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *Processor) ProcessFromClientAPI(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	// Allocate new log fields slice
 	fields := make([]kv.Field, 3, 4)
 	fields[0] = kv.Field{"activityType", cMsg.APActivityType}
 	fields[1] = kv.Field{"objectType", cMsg.APObjectType}
-	fields[2] = kv.Field{"fromAccount", cMsg.OriginAccount.Username}
+	fields[2] = kv.Field{"fromAccount", cMsg.Origin.Username}
 
 	// Include GTSModel in logs if appropriate.
 	if cMsg.GTSModel != nil &&
@@ -217,7 +202,7 @@ func (p *Processor) ProcessFromClientAPI(ctx context.Context, cMsg messages.From
 	return gtserror.Newf("unhandled: %s %s", cMsg.APActivityType, cMsg.APObjectType)
 }
 
-func (p *clientAPI) CreateAccount(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) CreateAccount(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	newUser, ok := cMsg.GTSModel.(*gtsmodel.User)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.User", cMsg.GTSModel)
@@ -241,14 +226,14 @@ func (p *clientAPI) CreateAccount(ctx context.Context, cMsg messages.FromClientA
 	return nil
 }
 
-func (p *clientAPI) CreateStatus(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) CreateStatus(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	status, ok := cMsg.GTSModel.(*gtsmodel.Status)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.Status", cMsg.GTSModel)
 	}
 
 	// Update stats for the actor account.
-	if err := p.utilF.incrementStatusesCount(ctx, cMsg.OriginAccount, status); err != nil {
+	if err := p.utils.incrementStatusesCount(ctx, cMsg.Origin, status); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
@@ -269,7 +254,7 @@ func (p *clientAPI) CreateStatus(ctx context.Context, cMsg messages.FromClientAP
 	return nil
 }
 
-func (p *clientAPI) CreatePollVote(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) CreatePollVote(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	// Cast the create poll vote attached to message.
 	vote, ok := cMsg.GTSModel.(*gtsmodel.PollVote)
 	if !ok {
@@ -310,14 +295,14 @@ func (p *clientAPI) CreatePollVote(ctx context.Context, cMsg messages.FromClient
 	return nil
 }
 
-func (p *clientAPI) CreateFollowReq(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) CreateFollowReq(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	followRequest, ok := cMsg.GTSModel.(*gtsmodel.FollowRequest)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.FollowRequest", cMsg.GTSModel)
 	}
 
 	// Update stats for the target account.
-	if err := p.utilF.incrementFollowRequestsCount(ctx, cMsg.TargetAccount); err != nil {
+	if err := p.utils.incrementFollowRequestsCount(ctx, cMsg.Target); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
@@ -338,7 +323,7 @@ func (p *clientAPI) CreateFollowReq(ctx context.Context, cMsg messages.FromClien
 	return nil
 }
 
-func (p *clientAPI) CreateLike(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) CreateLike(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	fave, ok := cMsg.GTSModel.(*gtsmodel.StatusFave)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.StatusFave", cMsg.GTSModel)
@@ -364,14 +349,14 @@ func (p *clientAPI) CreateLike(ctx context.Context, cMsg messages.FromClientAPI)
 	return nil
 }
 
-func (p *clientAPI) CreateAnnounce(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) CreateAnnounce(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	boost, ok := cMsg.GTSModel.(*gtsmodel.Status)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.Status", cMsg.GTSModel)
 	}
 
 	// Update stats for the actor account.
-	if err := p.utilF.incrementStatusesCount(ctx, cMsg.OriginAccount, boost); err != nil {
+	if err := p.utils.incrementStatusesCount(ctx, cMsg.Origin, boost); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
@@ -396,7 +381,7 @@ func (p *clientAPI) CreateAnnounce(ctx context.Context, cMsg messages.FromClient
 	return nil
 }
 
-func (p *clientAPI) CreateBlock(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) CreateBlock(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	block, ok := cMsg.GTSModel.(*gtsmodel.Block)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.Block", cMsg.GTSModel)
@@ -430,7 +415,7 @@ func (p *clientAPI) CreateBlock(ctx context.Context, cMsg messages.FromClientAPI
 	return nil
 }
 
-func (p *clientAPI) UpdateStatus(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) UpdateStatus(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	// Cast the updated Status model attached to msg.
 	status, ok := cMsg.GTSModel.(*gtsmodel.Status)
 	if !ok {
@@ -462,7 +447,7 @@ func (p *clientAPI) UpdateStatus(ctx context.Context, cMsg messages.FromClientAP
 	return nil
 }
 
-func (p *clientAPI) UpdateAccount(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) UpdateAccount(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	account, ok := cMsg.GTSModel.(*gtsmodel.Account)
 	if !ok {
 		return gtserror.Newf("cannot cast %T -> *gtsmodel.Account", cMsg.GTSModel)
@@ -475,7 +460,7 @@ func (p *clientAPI) UpdateAccount(ctx context.Context, cMsg messages.FromClientA
 	return nil
 }
 
-func (p *clientAPI) UpdateReport(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) UpdateReport(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	report, ok := cMsg.GTSModel.(*gtsmodel.Report)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.Report", cMsg.GTSModel)
@@ -494,23 +479,23 @@ func (p *clientAPI) UpdateReport(ctx context.Context, cMsg messages.FromClientAP
 	return nil
 }
 
-func (p *clientAPI) AcceptFollow(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) AcceptFollow(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	follow, ok := cMsg.GTSModel.(*gtsmodel.Follow)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.Follow", cMsg.GTSModel)
 	}
 
 	// Update stats for the target account.
-	if err := p.utilF.decrementFollowRequestsCount(ctx, cMsg.TargetAccount); err != nil {
+	if err := p.utils.decrementFollowRequestsCount(ctx, cMsg.Target); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
-	if err := p.utilF.incrementFollowersCount(ctx, cMsg.TargetAccount); err != nil {
+	if err := p.utils.incrementFollowersCount(ctx, cMsg.Target); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
 	// Update stats for the origin account.
-	if err := p.utilF.incrementFollowingCount(ctx, cMsg.OriginAccount); err != nil {
+	if err := p.utils.incrementFollowingCount(ctx, cMsg.Origin); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
@@ -525,14 +510,14 @@ func (p *clientAPI) AcceptFollow(ctx context.Context, cMsg messages.FromClientAP
 	return nil
 }
 
-func (p *clientAPI) RejectFollowRequest(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) RejectFollowRequest(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	followReq, ok := cMsg.GTSModel.(*gtsmodel.FollowRequest)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.FollowRequest", cMsg.GTSModel)
 	}
 
 	// Update stats for the target account.
-	if err := p.utilF.decrementFollowRequestsCount(ctx, cMsg.TargetAccount); err != nil {
+	if err := p.utils.decrementFollowRequestsCount(ctx, cMsg.Target); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
@@ -546,19 +531,19 @@ func (p *clientAPI) RejectFollowRequest(ctx context.Context, cMsg messages.FromC
 	return nil
 }
 
-func (p *clientAPI) UndoFollow(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) UndoFollow(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	follow, ok := cMsg.GTSModel.(*gtsmodel.Follow)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.Follow", cMsg.GTSModel)
 	}
 
 	// Update stats for the origin account.
-	if err := p.utilF.decrementFollowingCount(ctx, cMsg.OriginAccount); err != nil {
+	if err := p.utils.decrementFollowingCount(ctx, cMsg.Origin); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
 	// Update stats for the target account.
-	if err := p.utilF.decrementFollowersCount(ctx, cMsg.TargetAccount); err != nil {
+	if err := p.utils.decrementFollowersCount(ctx, cMsg.Target); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
@@ -569,7 +554,7 @@ func (p *clientAPI) UndoFollow(ctx context.Context, cMsg messages.FromClientAPI)
 	return nil
 }
 
-func (p *clientAPI) UndoBlock(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) UndoBlock(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	block, ok := cMsg.GTSModel.(*gtsmodel.Block)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.Block", cMsg.GTSModel)
@@ -582,7 +567,7 @@ func (p *clientAPI) UndoBlock(ctx context.Context, cMsg messages.FromClientAPI)
 	return nil
 }
 
-func (p *clientAPI) UndoFave(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) UndoFave(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	statusFave, ok := cMsg.GTSModel.(*gtsmodel.StatusFave)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.StatusFave", cMsg.GTSModel)
@@ -599,7 +584,7 @@ func (p *clientAPI) UndoFave(ctx context.Context, cMsg messages.FromClientAPI) e
 	return nil
 }
 
-func (p *clientAPI) UndoAnnounce(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) UndoAnnounce(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	status, ok := cMsg.GTSModel.(*gtsmodel.Status)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.Status", cMsg.GTSModel)
@@ -610,7 +595,7 @@ func (p *clientAPI) UndoAnnounce(ctx context.Context, cMsg messages.FromClientAP
 	}
 
 	// Update stats for the origin account.
-	if err := p.utilF.decrementStatusesCount(ctx, cMsg.OriginAccount); err != nil {
+	if err := p.utils.decrementStatusesCount(ctx, cMsg.Origin); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
@@ -629,7 +614,7 @@ func (p *clientAPI) UndoAnnounce(ctx context.Context, cMsg messages.FromClientAP
 	return nil
 }
 
-func (p *clientAPI) DeleteStatus(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) DeleteStatus(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	// Don't delete attachments, just unattach them:
 	// this request comes from the client API and the
 	// poster may want to use attachments again later.
@@ -648,12 +633,26 @@ func (p *clientAPI) DeleteStatus(ctx context.Context, cMsg messages.FromClientAP
 		return gtserror.Newf("db error populating status: %w", err)
 	}
 
-	if err := p.utilF.wipeStatus(ctx, status, deleteAttachments); err != nil {
+	// Drop any outgoing queued AP requests about / targeting
+	// this status, (stops queued likes, boosts, creates etc).
+	p.state.Workers.Delivery.Queue.Delete("ObjectID", status.URI)
+	p.state.Workers.Delivery.Queue.Delete("TargetID", status.URI)
+
+	// Drop any incoming queued client messages about / targeting
+	// status, (stops processing of local origin data for status).
+	p.state.Workers.Client.Queue.Delete("TargetURI", status.URI)
+
+	// Drop any incoming queued federator messages targeting status,
+	// (stops processing of remote origin data targeting this status).
+	p.state.Workers.Federator.Queue.Delete("TargetURI", status.URI)
+
+	// First perform the actual status deletion.
+	if err := p.utils.wipeStatus(ctx, status, deleteAttachments); err != nil {
 		log.Errorf(ctx, "error wiping status: %v", err)
 	}
 
 	// Update stats for the origin account.
-	if err := p.utilF.decrementStatusesCount(ctx, cMsg.OriginAccount); err != nil {
+	if err := p.utils.decrementStatusesCount(ctx, cMsg.Origin); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
@@ -670,7 +669,7 @@ func (p *clientAPI) DeleteStatus(ctx context.Context, cMsg messages.FromClientAP
 	return nil
 }
 
-func (p *clientAPI) DeleteAccount(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) DeleteAccount(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	// The originID of the delete, one of:
 	//   - ID of a domain block, for which
 	//     this account delete is a side effect.
@@ -684,21 +683,41 @@ func (p *clientAPI) DeleteAccount(ctx context.Context, cMsg messages.FromClientA
 	} else {
 		// Origin is whichever account
 		// originated this message.
-		originID = cMsg.OriginAccount.ID
+		originID = cMsg.Origin.ID
 	}
 
-	if err := p.federate.DeleteAccount(ctx, cMsg.TargetAccount); err != nil {
+	// Extract target account.
+	account := cMsg.Target
+
+	// Drop any outgoing queued AP requests to / from / targeting
+	// this account, (stops queued likes, boosts, creates etc).
+	p.state.Workers.Delivery.Queue.Delete("ActorID", account.URI)
+	p.state.Workers.Delivery.Queue.Delete("ObjectID", account.URI)
+	p.state.Workers.Delivery.Queue.Delete("TargetID", account.URI)
+
+	// Drop any incoming queued client messages to / from this
+	// account, (stops processing of local origin data for acccount).
+	p.state.Workers.Client.Queue.Delete("Origin.ID", account.ID)
+	p.state.Workers.Client.Queue.Delete("Target.ID", account.ID)
+	p.state.Workers.Client.Queue.Delete("TargetURI", account.URI)
+
+	// Drop any incoming queued federator messages to this account,
+	// (stops processing of remote origin data targeting this account).
+	p.state.Workers.Federator.Queue.Delete("Receiving.ID", account.ID)
+	p.state.Workers.Federator.Queue.Delete("TargetURI", account.URI)
+
+	if err := p.federate.DeleteAccount(ctx, cMsg.Target); err != nil {
 		log.Errorf(ctx, "error federating account delete: %v", err)
 	}
 
-	if err := p.account.Delete(ctx, cMsg.TargetAccount, originID); err != nil {
+	if err := p.account.Delete(ctx, cMsg.Target, originID); err != nil {
 		log.Errorf(ctx, "error deleting account: %v", err)
 	}
 
 	return nil
 }
 
-func (p *clientAPI) ReportAccount(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) ReportAccount(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	report, ok := cMsg.GTSModel.(*gtsmodel.Report)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.Report", cMsg.GTSModel)
@@ -719,28 +738,28 @@ func (p *clientAPI) ReportAccount(ctx context.Context, cMsg messages.FromClientA
 	return nil
 }
 
-func (p *clientAPI) MoveAccount(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) MoveAccount(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	// Redirect each local follower of
 	// OriginAccount to follow move target.
-	p.utilF.redirectFollowers(ctx, cMsg.OriginAccount, cMsg.TargetAccount)
+	p.utils.redirectFollowers(ctx, cMsg.Origin, cMsg.Target)
 
 	// At this point, we know OriginAccount has the
 	// Move set on it. Just make sure it's populated.
-	if err := p.state.DB.PopulateMove(ctx, cMsg.OriginAccount.Move); err != nil {
+	if err := p.state.DB.PopulateMove(ctx, cMsg.Origin.Move); err != nil {
 		return gtserror.Newf("error populating Move: %w", err)
 	}
 
 	// Now send the Move message out to
 	// OriginAccount's (remote) followers.
-	if err := p.federate.MoveAccount(ctx, cMsg.OriginAccount); err != nil {
+	if err := p.federate.MoveAccount(ctx, cMsg.Origin); err != nil {
 		return gtserror.Newf("error federating account move: %w", err)
 	}
 
 	// Mark the move attempt as successful.
-	cMsg.OriginAccount.Move.SucceededAt = cMsg.OriginAccount.Move.AttemptedAt
+	cMsg.Origin.Move.SucceededAt = cMsg.Origin.Move.AttemptedAt
 	if err := p.state.DB.UpdateMove(
 		ctx,
-		cMsg.OriginAccount.Move,
+		cMsg.Origin.Move,
 		"succeeded_at",
 	); err != nil {
 		return gtserror.Newf("error marking move as successful: %w", err)
@@ -749,7 +768,7 @@ func (p *clientAPI) MoveAccount(ctx context.Context, cMsg messages.FromClientAPI
 	return nil
 }
 
-func (p *clientAPI) AcceptAccount(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) AcceptAccount(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	newUser, ok := cMsg.GTSModel.(*gtsmodel.User)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.User", cMsg.GTSModel)
@@ -772,17 +791,17 @@ func (p *clientAPI) AcceptAccount(ctx context.Context, cMsg messages.FromClientA
 	return nil
 }
 
-func (p *clientAPI) RejectAccount(ctx context.Context, cMsg messages.FromClientAPI) error {
+func (p *clientAPI) RejectAccount(ctx context.Context, cMsg *messages.FromClientAPI) error {
 	deniedUser, ok := cMsg.GTSModel.(*gtsmodel.DeniedUser)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.DeniedUser", cMsg.GTSModel)
 	}
 
 	// Remove the account.
-	if err := p.state.DB.DeleteAccount(ctx, cMsg.TargetAccount.ID); err != nil {
+	if err := p.state.DB.DeleteAccount(ctx, cMsg.Target.ID); err != nil {
 		log.Errorf(ctx,
 			"db error deleting account %s: %v",
-			cMsg.TargetAccount.ID, err,
+			cMsg.Target.ID, err,
 		)
 	}
 
diff --git a/internal/processing/workers/fromclientapi_test.go b/internal/processing/workers/fromclientapi_test.go
index 5e294597d..3aa0fbdad 100644
--- a/internal/processing/workers/fromclientapi_test.go
+++ b/internal/processing/workers/fromclientapi_test.go
@@ -197,11 +197,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusWithNotification() {
 	// Process the new status.
 	if err := suite.processor.Workers().ProcessFromClientAPI(
 		ctx,
-		messages.FromClientAPI{
+		&messages.FromClientAPI{
 			APObjectType:   ap.ObjectNote,
 			APActivityType: ap.ActivityCreate,
 			GTSModel:       status,
-			OriginAccount:  postingAccount,
+			Origin:         postingAccount,
 		},
 	); err != nil {
 		suite.FailNow(err.Error())
@@ -291,11 +291,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusReply() {
 	// Process the new status.
 	if err := suite.processor.Workers().ProcessFromClientAPI(
 		ctx,
-		messages.FromClientAPI{
+		&messages.FromClientAPI{
 			APObjectType:   ap.ObjectNote,
 			APActivityType: ap.ActivityCreate,
 			GTSModel:       status,
-			OriginAccount:  postingAccount,
+			Origin:         postingAccount,
 		},
 	); err != nil {
 		suite.FailNow(err.Error())
@@ -355,11 +355,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusReplyMuted() {
 	// Process the new status.
 	if err := suite.processor.Workers().ProcessFromClientAPI(
 		ctx,
-		messages.FromClientAPI{
+		&messages.FromClientAPI{
 			APObjectType:   ap.ObjectNote,
 			APActivityType: ap.ActivityCreate,
 			GTSModel:       status,
-			OriginAccount:  postingAccount,
+			Origin:         postingAccount,
 		},
 	); err != nil {
 		suite.FailNow(err.Error())
@@ -409,11 +409,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusBoostMuted() {
 	// Process the new status.
 	if err := suite.processor.Workers().ProcessFromClientAPI(
 		ctx,
-		messages.FromClientAPI{
+		&messages.FromClientAPI{
 			APObjectType:   ap.ActivityAnnounce,
 			APActivityType: ap.ActivityCreate,
 			GTSModel:       status,
-			OriginAccount:  postingAccount,
+			Origin:         postingAccount,
 		},
 	); err != nil {
 		suite.FailNow(err.Error())
@@ -467,11 +467,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusListRepliesPolicyLis
 	// Process the new status.
 	if err := suite.processor.Workers().ProcessFromClientAPI(
 		ctx,
-		messages.FromClientAPI{
+		&messages.FromClientAPI{
 			APObjectType:   ap.ObjectNote,
 			APActivityType: ap.ActivityCreate,
 			GTSModel:       status,
-			OriginAccount:  postingAccount,
+			Origin:         postingAccount,
 		},
 	); err != nil {
 		suite.FailNow(err.Error())
@@ -540,11 +540,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusListRepliesPolicyLis
 	// Process the new status.
 	if err := suite.processor.Workers().ProcessFromClientAPI(
 		ctx,
-		messages.FromClientAPI{
+		&messages.FromClientAPI{
 			APObjectType:   ap.ObjectNote,
 			APActivityType: ap.ActivityCreate,
 			GTSModel:       status,
-			OriginAccount:  postingAccount,
+			Origin:         postingAccount,
 		},
 	); err != nil {
 		suite.FailNow(err.Error())
@@ -608,11 +608,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusReplyListRepliesPoli
 	// Process the new status.
 	if err := suite.processor.Workers().ProcessFromClientAPI(
 		ctx,
-		messages.FromClientAPI{
+		&messages.FromClientAPI{
 			APObjectType:   ap.ObjectNote,
 			APActivityType: ap.ActivityCreate,
 			GTSModel:       status,
-			OriginAccount:  postingAccount,
+			Origin:         postingAccount,
 		},
 	); err != nil {
 		suite.FailNow(err.Error())
@@ -664,11 +664,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusBoost() {
 	// Process the new status.
 	if err := suite.processor.Workers().ProcessFromClientAPI(
 		ctx,
-		messages.FromClientAPI{
+		&messages.FromClientAPI{
 			APObjectType:   ap.ActivityAnnounce,
 			APActivityType: ap.ActivityCreate,
 			GTSModel:       status,
-			OriginAccount:  postingAccount,
+			Origin:         postingAccount,
 		},
 	); err != nil {
 		suite.FailNow(err.Error())
@@ -729,11 +729,11 @@ func (suite *FromClientAPITestSuite) TestProcessCreateStatusBoostNoReblogs() {
 	// Process the new status.
 	if err := suite.processor.Workers().ProcessFromClientAPI(
 		ctx,
-		messages.FromClientAPI{
+		&messages.FromClientAPI{
 			APObjectType:   ap.ActivityAnnounce,
 			APActivityType: ap.ActivityCreate,
 			GTSModel:       status,
-			OriginAccount:  postingAccount,
+			Origin:         postingAccount,
 		},
 	); err != nil {
 		suite.FailNow(err.Error())
@@ -776,11 +776,11 @@ func (suite *FromClientAPITestSuite) TestProcessStatusDelete() {
 	// Process the status delete.
 	if err := suite.processor.Workers().ProcessFromClientAPI(
 		ctx,
-		messages.FromClientAPI{
+		&messages.FromClientAPI{
 			APObjectType:   ap.ObjectNote,
 			APActivityType: ap.ActivityDelete,
 			GTSModel:       deletedStatus,
-			OriginAccount:  deletingAccount,
+			Origin:         deletingAccount,
 		},
 	); err != nil {
 		suite.FailNow(err.Error())
diff --git a/internal/processing/workers/fromfediapi.go b/internal/processing/workers/fromfediapi.go
index 0b1106a9e..fcd5b38f2 100644
--- a/internal/processing/workers/fromfediapi.go
+++ b/internal/processing/workers/fromfediapi.go
@@ -19,13 +19,14 @@ package workers
 
 import (
 	"context"
+	"errors"
 
 	"codeberg.org/gruf/go-kv"
 	"codeberg.org/gruf/go-logger/v2/level"
 	"github.com/superseriousbusiness/gotosocial/internal/ap"
+	"github.com/superseriousbusiness/gotosocial/internal/db"
 	"github.com/superseriousbusiness/gotosocial/internal/federation/dereferencing"
 
-	"github.com/superseriousbusiness/gotosocial/internal/gtscontext"
 	"github.com/superseriousbusiness/gotosocial/internal/gtserror"
 	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
 	"github.com/superseriousbusiness/gotosocial/internal/log"
@@ -43,34 +44,20 @@ type fediAPI struct {
 	surface  *surface
 	federate *federate
 	account  *account.Processor
-	utilF    *utilF
+	utils    *utils
 }
 
-func (p *Processor) EnqueueFediAPI(cctx context.Context, msgs ...messages.FromFediAPI) {
-	_ = p.workers.Federator.MustEnqueueCtx(cctx, func(wctx context.Context) {
-		// Copy caller ctx values to worker's.
-		wctx = gtscontext.WithValues(wctx, cctx)
-
-		// Process worker messages.
-		for _, msg := range msgs {
-			if err := p.ProcessFromFediAPI(wctx, msg); err != nil {
-				log.Errorf(wctx, "error processing fedi API message: %v", err)
-			}
-		}
-	})
-}
-
-func (p *Processor) ProcessFromFediAPI(ctx context.Context, fMsg messages.FromFediAPI) error {
+func (p *Processor) ProcessFromFediAPI(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	// Allocate new log fields slice
 	fields := make([]kv.Field, 3, 5)
 	fields[0] = kv.Field{"activityType", fMsg.APActivityType}
 	fields[1] = kv.Field{"objectType", fMsg.APObjectType}
-	fields[2] = kv.Field{"toAccount", fMsg.ReceivingAccount.Username}
+	fields[2] = kv.Field{"toAccount", fMsg.Receiving.Username}
 
-	if fMsg.APIri != nil {
+	if fMsg.APIRI != nil {
 		// An IRI was supplied, append to log
 		fields = append(fields, kv.Field{
-			"iri", fMsg.APIri,
+			"iri", fMsg.APIRI,
 		})
 	}
 
@@ -168,7 +155,7 @@ func (p *Processor) ProcessFromFediAPI(ctx context.Context, fMsg messages.FromFe
 	return gtserror.Newf("unhandled: %s %s", fMsg.APActivityType, fMsg.APObjectType)
 }
 
-func (p *fediAPI) CreateStatus(ctx context.Context, fMsg messages.FromFediAPI) error {
+func (p *fediAPI) CreateStatus(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	var (
 		status     *gtsmodel.Status
 		statusable ap.Statusable
@@ -178,11 +165,11 @@ func (p *fediAPI) CreateStatus(ctx context.Context, fMsg messages.FromFediAPI) e
 	var ok bool
 
 	switch {
-	case fMsg.APObjectModel != nil:
+	case fMsg.APObject != nil:
 		// A model was provided, extract this from message.
-		statusable, ok = fMsg.APObjectModel.(ap.Statusable)
+		statusable, ok = fMsg.APObject.(ap.Statusable)
 		if !ok {
-			return gtserror.Newf("cannot cast %T -> ap.Statusable", fMsg.APObjectModel)
+			return gtserror.Newf("cannot cast %T -> ap.Statusable", fMsg.APObject)
 		}
 
 		// Create bare-bones model to pass
@@ -196,7 +183,7 @@ func (p *fediAPI) CreateStatus(ctx context.Context, fMsg messages.FromFediAPI) e
 		// statusable model, which it will use to further flesh out
 		// the bare bones model and insert it into the database.
 		status, statusable, err = p.federate.RefreshStatus(ctx,
-			fMsg.ReceivingAccount.Username,
+			fMsg.Receiving.Username,
 			bareStatus,
 			statusable,
 			// Force refresh within 5min window.
@@ -206,15 +193,15 @@ func (p *fediAPI) CreateStatus(ctx context.Context, fMsg messages.FromFediAPI) e
 			return gtserror.Newf("error processing new status %s: %w", bareStatus.URI, err)
 		}
 
-	case fMsg.APIri != nil:
+	case fMsg.APIRI != nil:
 		// Model was not set, deref with IRI (this is a forward).
 		// This will also cause the status to be inserted into the db.
 		status, statusable, err = p.federate.GetStatusByURI(ctx,
-			fMsg.ReceivingAccount.Username,
-			fMsg.APIri,
+			fMsg.Receiving.Username,
+			fMsg.APIRI,
 		)
 		if err != nil {
-			return gtserror.Newf("error dereferencing forwarded status %s: %w", fMsg.APIri, err)
+			return gtserror.Newf("error dereferencing forwarded status %s: %w", fMsg.APIRI, err)
 		}
 
 	default:
@@ -230,7 +217,7 @@ func (p *fediAPI) CreateStatus(ctx context.Context, fMsg messages.FromFediAPI) e
 	}
 
 	// Update stats for the remote account.
-	if err := p.utilF.incrementStatusesCount(ctx, fMsg.RequestingAccount, status); err != nil {
+	if err := p.utils.incrementStatusesCount(ctx, fMsg.Requesting, status); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
@@ -248,7 +235,7 @@ func (p *fediAPI) CreateStatus(ctx context.Context, fMsg messages.FromFediAPI) e
 	return nil
 }
 
-func (p *fediAPI) CreatePollVote(ctx context.Context, fMsg messages.FromFediAPI) error {
+func (p *fediAPI) CreatePollVote(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	// Cast poll vote type from the worker message.
 	vote, ok := fMsg.GTSModel.(*gtsmodel.PollVote)
 	if !ok {
@@ -293,7 +280,7 @@ func (p *fediAPI) CreatePollVote(ctx context.Context, fMsg messages.FromFediAPI)
 	return nil
 }
 
-func (p *fediAPI) CreateFollowReq(ctx context.Context, fMsg messages.FromFediAPI) error {
+func (p *fediAPI) CreateFollowReq(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	followRequest, ok := fMsg.GTSModel.(*gtsmodel.FollowRequest)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.FollowRequest", fMsg.GTSModel)
@@ -310,7 +297,7 @@ func (p *fediAPI) CreateFollowReq(ctx context.Context, fMsg messages.FromFediAPI
 		}
 
 		// And update stats for the local account.
-		if err := p.utilF.incrementFollowRequestsCount(ctx, fMsg.ReceivingAccount); err != nil {
+		if err := p.utils.incrementFollowRequestsCount(ctx, fMsg.Receiving); err != nil {
 			log.Errorf(ctx, "error updating account stats: %v", err)
 		}
 
@@ -330,12 +317,12 @@ func (p *fediAPI) CreateFollowReq(ctx context.Context, fMsg messages.FromFediAPI
 	}
 
 	// Update stats for the local account.
-	if err := p.utilF.incrementFollowersCount(ctx, fMsg.ReceivingAccount); err != nil {
+	if err := p.utils.incrementFollowersCount(ctx, fMsg.Receiving); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
 	// Update stats for the remote account.
-	if err := p.utilF.incrementFollowingCount(ctx, fMsg.RequestingAccount); err != nil {
+	if err := p.utils.incrementFollowingCount(ctx, fMsg.Requesting); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
@@ -350,7 +337,7 @@ func (p *fediAPI) CreateFollowReq(ctx context.Context, fMsg messages.FromFediAPI
 	return nil
 }
 
-func (p *fediAPI) CreateLike(ctx context.Context, fMsg messages.FromFediAPI) error {
+func (p *fediAPI) CreateLike(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	fave, ok := fMsg.GTSModel.(*gtsmodel.StatusFave)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.StatusFave", fMsg.GTSModel)
@@ -372,7 +359,7 @@ func (p *fediAPI) CreateLike(ctx context.Context, fMsg messages.FromFediAPI) err
 	return nil
 }
 
-func (p *fediAPI) CreateAnnounce(ctx context.Context, fMsg messages.FromFediAPI) error {
+func (p *fediAPI) CreateAnnounce(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	boost, ok := fMsg.GTSModel.(*gtsmodel.Status)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.Status", fMsg.GTSModel)
@@ -386,7 +373,7 @@ func (p *fediAPI) CreateAnnounce(ctx context.Context, fMsg messages.FromFediAPI)
 	boost, err = p.federate.EnrichAnnounce(
 		ctx,
 		boost,
-		fMsg.ReceivingAccount.Username,
+		fMsg.Receiving.Username,
 	)
 	if err != nil {
 		if gtserror.IsUnretrievable(err) {
@@ -400,7 +387,7 @@ func (p *fediAPI) CreateAnnounce(ctx context.Context, fMsg messages.FromFediAPI)
 	}
 
 	// Update stats for the remote account.
-	if err := p.utilF.incrementStatusesCount(ctx, fMsg.RequestingAccount, boost); err != nil {
+	if err := p.utils.incrementStatusesCount(ctx, fMsg.Requesting, boost); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
@@ -420,7 +407,7 @@ func (p *fediAPI) CreateAnnounce(ctx context.Context, fMsg messages.FromFediAPI)
 	return nil
 }
 
-func (p *fediAPI) CreateBlock(ctx context.Context, fMsg messages.FromFediAPI) error {
+func (p *fediAPI) CreateBlock(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	block, ok := fMsg.GTSModel.(*gtsmodel.Block)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.Block", fMsg.GTSModel)
@@ -499,7 +486,7 @@ func (p *fediAPI) CreateBlock(ctx context.Context, fMsg messages.FromFediAPI) er
 	return nil
 }
 
-func (p *fediAPI) CreateFlag(ctx context.Context, fMsg messages.FromFediAPI) error {
+func (p *fediAPI) CreateFlag(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	incomingReport, ok := fMsg.GTSModel.(*gtsmodel.Report)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.Report", fMsg.GTSModel)
@@ -515,7 +502,7 @@ func (p *fediAPI) CreateFlag(ctx context.Context, fMsg messages.FromFediAPI) err
 	return nil
 }
 
-func (p *fediAPI) UpdateAccount(ctx context.Context, fMsg messages.FromFediAPI) error {
+func (p *fediAPI) UpdateAccount(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	// Parse the old/existing account model.
 	account, ok := fMsg.GTSModel.(*gtsmodel.Account)
 	if !ok {
@@ -523,15 +510,15 @@ func (p *fediAPI) UpdateAccount(ctx context.Context, fMsg messages.FromFediAPI)
 	}
 
 	// Because this was an Update, the new Accountable should be set on the message.
-	apubAcc, ok := fMsg.APObjectModel.(ap.Accountable)
+	apubAcc, ok := fMsg.APObject.(ap.Accountable)
 	if !ok {
-		return gtserror.Newf("cannot cast %T -> ap.Accountable", fMsg.APObjectModel)
+		return gtserror.Newf("cannot cast %T -> ap.Accountable", fMsg.APObject)
 	}
 
 	// Fetch up-to-date bio, avatar, header, etc.
 	_, _, err := p.federate.RefreshAccount(
 		ctx,
-		fMsg.ReceivingAccount.Username,
+		fMsg.Receiving.Username,
 		account,
 		apubAcc,
 		// Force refresh within 5min window.
@@ -544,25 +531,25 @@ func (p *fediAPI) UpdateAccount(ctx context.Context, fMsg messages.FromFediAPI)
 	return nil
 }
 
-func (p *fediAPI) AcceptFollow(ctx context.Context, fMsg messages.FromFediAPI) error {
+func (p *fediAPI) AcceptFollow(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	// Update stats for the remote account.
-	if err := p.utilF.decrementFollowRequestsCount(ctx, fMsg.RequestingAccount); err != nil {
+	if err := p.utils.decrementFollowRequestsCount(ctx, fMsg.Requesting); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
-	if err := p.utilF.incrementFollowersCount(ctx, fMsg.RequestingAccount); err != nil {
+	if err := p.utils.incrementFollowersCount(ctx, fMsg.Requesting); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
 	// Update stats for the local account.
-	if err := p.utilF.incrementFollowingCount(ctx, fMsg.ReceivingAccount); err != nil {
+	if err := p.utils.incrementFollowingCount(ctx, fMsg.Receiving); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
 	return nil
 }
 
-func (p *fediAPI) UpdateStatus(ctx context.Context, fMsg messages.FromFediAPI) error {
+func (p *fediAPI) UpdateStatus(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	// Cast the existing Status model attached to msg.
 	existing, ok := fMsg.GTSModel.(*gtsmodel.Status)
 	if !ok {
@@ -570,12 +557,12 @@ func (p *fediAPI) UpdateStatus(ctx context.Context, fMsg messages.FromFediAPI) e
 	}
 
 	// Cast the updated ActivityPub statusable object .
-	apStatus, _ := fMsg.APObjectModel.(ap.Statusable)
+	apStatus, _ := fMsg.APObject.(ap.Statusable)
 
 	// Fetch up-to-date attach status attachments, etc.
 	status, _, err := p.federate.RefreshStatus(
 		ctx,
-		fMsg.ReceivingAccount.Username,
+		fMsg.Receiving.Username,
 		existing,
 		apStatus,
 		// Force refresh within 5min window.
@@ -605,7 +592,7 @@ func (p *fediAPI) UpdateStatus(ctx context.Context, fMsg messages.FromFediAPI) e
 	return nil
 }
 
-func (p *fediAPI) DeleteStatus(ctx context.Context, fMsg messages.FromFediAPI) error {
+func (p *fediAPI) DeleteStatus(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	// Delete attachments from this status, since this request
 	// comes from the federating API, and there's no way the
 	// poster can do a delete + redraft for it on our instance.
@@ -616,12 +603,34 @@ func (p *fediAPI) DeleteStatus(ctx context.Context, fMsg messages.FromFediAPI) e
 		return gtserror.Newf("%T not parseable as *gtsmodel.Status", fMsg.GTSModel)
 	}
 
-	if err := p.utilF.wipeStatus(ctx, status, deleteAttachments); err != nil {
+	// Try to populate status structs if possible,
+	// in order to more thoroughly remove them.
+	if err := p.state.DB.PopulateStatus(
+		ctx, status,
+	); err != nil && !errors.Is(err, db.ErrNoEntries) {
+		return gtserror.Newf("db error populating status: %w", err)
+	}
+
+	// Drop any outgoing queued AP requests about / targeting
+	// this status, (stops queued likes, boosts, creates etc).
+	p.state.Workers.Delivery.Queue.Delete("ObjectID", status.URI)
+	p.state.Workers.Delivery.Queue.Delete("TargetID", status.URI)
+
+	// Drop any incoming queued client messages about / targeting
+	// status, (stops processing of local origin data for status).
+	p.state.Workers.Client.Queue.Delete("TargetURI", status.URI)
+
+	// Drop any incoming queued federator messages targeting status,
+	// (stops processing of remote origin data targeting this status).
+	p.state.Workers.Federator.Queue.Delete("TargetURI", status.URI)
+
+	// First perform the actual status deletion.
+	if err := p.utils.wipeStatus(ctx, status, deleteAttachments); err != nil {
 		log.Errorf(ctx, "error wiping status: %v", err)
 	}
 
 	// Update stats for the remote account.
-	if err := p.utilF.decrementStatusesCount(ctx, fMsg.RequestingAccount); err != nil {
+	if err := p.utils.decrementStatusesCount(ctx, fMsg.Requesting); err != nil {
 		log.Errorf(ctx, "error updating account stats: %v", err)
 	}
 
@@ -634,12 +643,28 @@ func (p *fediAPI) DeleteStatus(ctx context.Context, fMsg messages.FromFediAPI) e
 	return nil
 }
 
-func (p *fediAPI) DeleteAccount(ctx context.Context, fMsg messages.FromFediAPI) error {
+func (p *fediAPI) DeleteAccount(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	account, ok := fMsg.GTSModel.(*gtsmodel.Account)
 	if !ok {
 		return gtserror.Newf("%T not parseable as *gtsmodel.Account", fMsg.GTSModel)
 	}
 
+	// Drop any outgoing queued AP requests to / from / targeting
+	// this account, (stops queued likes, boosts, creates etc).
+	p.state.Workers.Delivery.Queue.Delete("ObjectID", account.URI)
+	p.state.Workers.Delivery.Queue.Delete("TargetID", account.URI)
+
+	// Drop any incoming queued client messages to / from this
+	// account, (stops processing of local origin data for acccount).
+	p.state.Workers.Client.Queue.Delete("Target.ID", account.ID)
+	p.state.Workers.Client.Queue.Delete("TargetURI", account.URI)
+
+	// Drop any incoming queued federator messages to this account,
+	// (stops processing of remote origin data targeting this account).
+	p.state.Workers.Federator.Queue.Delete("Requesting.ID", account.ID)
+	p.state.Workers.Federator.Queue.Delete("TargetURI", account.URI)
+
+	// First perform the actual account deletion.
 	if err := p.account.Delete(ctx, account, account.ID); err != nil {
 		log.Errorf(ctx, "error deleting account: %v", err)
 	}
diff --git a/internal/processing/workers/fromfediapi_move.go b/internal/processing/workers/fromfediapi_move.go
index 0188a5d14..d1e43c0c7 100644
--- a/internal/processing/workers/fromfediapi_move.go
+++ b/internal/processing/workers/fromfediapi_move.go
@@ -220,10 +220,7 @@ func (p *fediAPI) GetOrCreateMove(
 //	APActivityType:   "Move"
 //	GTSModel:         stub *gtsmodel.Move.
 //	ReceivingAccount: Account of inbox owner receiving the Move.
-func (p *fediAPI) MoveAccount(ctx context.Context, fMsg messages.FromFediAPI) error {
-	// The account who received the Move message.
-	receiver := fMsg.ReceivingAccount
-
+func (p *fediAPI) MoveAccount(ctx context.Context, fMsg *messages.FromFediAPI) error {
 	// *gtsmodel.Move activity.
 	stubMove, ok := fMsg.GTSModel.(*gtsmodel.Move)
 	if !ok {
@@ -236,7 +233,7 @@ func (p *fediAPI) MoveAccount(ctx context.Context, fMsg messages.FromFediAPI) er
 	// Move origin and target info.
 	var (
 		originAcctURIStr = stubMove.OriginURI
-		originAcct       = fMsg.RequestingAccount
+		originAcct       = fMsg.Requesting
 		targetAcctURIStr = stubMove.TargetURI
 		targetAcctURI    = stubMove.Target
 	)
@@ -308,7 +305,7 @@ func (p *fediAPI) MoveAccount(ctx context.Context, fMsg messages.FromFediAPI) er
 	// Account to which the Move is taking place.
 	targetAcct, targetAcctable, err := p.federate.GetAccountByURI(
 		ctx,
-		receiver.Username,
+		fMsg.Receiving.Username,
 		targetAcctURI,
 	)
 	if err != nil {
@@ -340,7 +337,7 @@ func (p *fediAPI) MoveAccount(ctx context.Context, fMsg messages.FromFediAPI) er
 		// Force refresh Move target account
 		// to ensure we have up-to-date version.
 		targetAcct, _, err = p.federate.RefreshAccount(ctx,
-			receiver.Username,
+			fMsg.Receiving.Username,
 			targetAcct,
 			targetAcctable,
 			dereferencing.Freshest,
@@ -379,7 +376,7 @@ func (p *fediAPI) MoveAccount(ctx context.Context, fMsg messages.FromFediAPI) er
 
 	// Transfer originAcct's followers
 	// on this instance to targetAcct.
-	redirectOK := p.utilF.redirectFollowers(
+	redirectOK := p.utils.redirectFollowers(
 		ctx,
 		originAcct,
 		targetAcct,
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)
 
diff --git a/internal/processing/workers/util.go b/internal/processing/workers/util.go
index cd936f428..a01982e1a 100644
--- a/internal/processing/workers/util.go
+++ b/internal/processing/workers/util.go
@@ -32,9 +32,9 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/state"
 )
 
-// utilF wraps util functions used by both
+// util provides util functions used by both
 // the fromClientAPI and fromFediAPI functions.
-type utilF struct {
+type utils struct {
 	state   *state.State
 	media   *media.Processor
 	account *account.Processor
@@ -45,7 +45,7 @@ type utilF struct {
 // used to totally delete a status + all
 // its attachments, notifications, boosts,
 // and timeline entries.
-func (u *utilF) wipeStatus(
+func (u *utils) wipeStatus(
 	ctx context.Context,
 	statusToDelete *gtsmodel.Status,
 	deleteAttachments bool,
@@ -152,7 +152,7 @@ func (u *utilF) wipeStatus(
 // already, and the Move must be valid.
 //
 // Return bool will be true if all goes OK.
-func (u *utilF) redirectFollowers(
+func (u *utils) redirectFollowers(
 	ctx context.Context,
 	originAcct *gtsmodel.Account,
 	targetAcct *gtsmodel.Account,
@@ -239,7 +239,7 @@ func (u *utilF) redirectFollowers(
 	return true
 }
 
-func (u *utilF) incrementStatusesCount(
+func (u *utils) incrementStatusesCount(
 	ctx context.Context,
 	account *gtsmodel.Account,
 	status *gtsmodel.Status,
@@ -271,7 +271,7 @@ func (u *utilF) incrementStatusesCount(
 	return nil
 }
 
-func (u *utilF) decrementStatusesCount(
+func (u *utils) decrementStatusesCount(
 	ctx context.Context,
 	account *gtsmodel.Account,
 ) error {
@@ -305,7 +305,7 @@ func (u *utilF) decrementStatusesCount(
 	return nil
 }
 
-func (u *utilF) incrementFollowersCount(
+func (u *utils) incrementFollowersCount(
 	ctx context.Context,
 	account *gtsmodel.Account,
 ) error {
@@ -334,7 +334,7 @@ func (u *utilF) incrementFollowersCount(
 	return nil
 }
 
-func (u *utilF) decrementFollowersCount(
+func (u *utils) decrementFollowersCount(
 	ctx context.Context,
 	account *gtsmodel.Account,
 ) error {
@@ -368,7 +368,7 @@ func (u *utilF) decrementFollowersCount(
 	return nil
 }
 
-func (u *utilF) incrementFollowingCount(
+func (u *utils) incrementFollowingCount(
 	ctx context.Context,
 	account *gtsmodel.Account,
 ) error {
@@ -397,7 +397,7 @@ func (u *utilF) incrementFollowingCount(
 	return nil
 }
 
-func (u *utilF) decrementFollowingCount(
+func (u *utils) decrementFollowingCount(
 	ctx context.Context,
 	account *gtsmodel.Account,
 ) error {
@@ -431,7 +431,7 @@ func (u *utilF) decrementFollowingCount(
 	return nil
 }
 
-func (u *utilF) incrementFollowRequestsCount(
+func (u *utils) incrementFollowRequestsCount(
 	ctx context.Context,
 	account *gtsmodel.Account,
 ) error {
@@ -460,7 +460,7 @@ func (u *utilF) incrementFollowRequestsCount(
 	return nil
 }
 
-func (u *utilF) decrementFollowRequestsCount(
+func (u *utils) decrementFollowRequestsCount(
 	ctx context.Context,
 	account *gtsmodel.Account,
 ) error {
diff --git a/internal/processing/workers/workers.go b/internal/processing/workers/workers.go
index 8488e501c..1159b61a5 100644
--- a/internal/processing/workers/workers.go
+++ b/internal/processing/workers/workers.go
@@ -30,9 +30,9 @@ import (
 )
 
 type Processor struct {
+	clientAPI clientAPI
+	fediAPI   fediAPI
 	workers   *workers.Workers
-	clientAPI *clientAPI
-	fediAPI   *fediAPI
 }
 
 func New(
@@ -45,6 +45,14 @@ func New(
 	media *media.Processor,
 	stream *stream.Processor,
 ) Processor {
+	// Init federate logic
+	// wrapper struct.
+	federate := &federate{
+		Federator: federator,
+		state:     state,
+		converter: converter,
+	}
+
 	// Init surface logic
 	// wrapper struct.
 	surface := &surface{
@@ -55,16 +63,8 @@ func New(
 		emailSender: emailSender,
 	}
 
-	// Init federate logic
-	// wrapper struct.
-	federate := &federate{
-		Federator: federator,
-		state:     state,
-		converter: converter,
-	}
-
 	// Init shared util funcs.
-	utilF := &utilF{
+	utils := &utils{
 		state:   state,
 		media:   media,
 		account: account,
@@ -73,20 +73,20 @@ func New(
 
 	return Processor{
 		workers: &state.Workers,
-		clientAPI: &clientAPI{
+		clientAPI: clientAPI{
 			state:     state,
 			converter: converter,
 			surface:   surface,
 			federate:  federate,
 			account:   account,
-			utilF:     utilF,
+			utils:     utils,
 		},
-		fediAPI: &fediAPI{
+		fediAPI: fediAPI{
 			state:    state,
 			surface:  surface,
 			federate: federate,
 			account:  account,
-			utilF:    utilF,
+			utils:    utils,
 		},
 	}
 }
diff --git a/internal/processing/workers/workers_test.go b/internal/processing/workers/workers_test.go
index 3ab19d38d..af1428e64 100644
--- a/internal/processing/workers/workers_test.go
+++ b/internal/processing/workers/workers_test.go
@@ -127,9 +127,6 @@ func (suite *WorkersTestSuite) SetupTest() {
 	suite.processor = processing.NewProcessor(cleaner.New(&suite.state), suite.typeconverter, suite.federator, suite.oauthServer, suite.mediaManager, &suite.state, suite.emailSender)
 	testrig.StartWorkers(&suite.state, suite.processor.Workers())
 
-	suite.state.Workers.EnqueueClientAPI = suite.processor.Workers().EnqueueClientAPI
-	suite.state.Workers.EnqueueFediAPI = suite.processor.Workers().EnqueueFediAPI
-
 	testrig.StandardDBSetup(suite.db, suite.testAccounts)
 	testrig.StandardStorageSetup(suite.storage, "../../../testrig/media")
 }