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

diff --git a/internal/federation/federatingdb/delete.go b/internal/federation/federatingdb/delete.go
index 622ef6d3d..7e9b66c5a 100644
--- a/internal/federation/federatingdb/delete.go
+++ b/internal/federation/federatingdb/delete.go
@@ -19,10 +19,13 @@ package federatingdb
 
 import (
 	"context"
+	"errors"
 	"net/url"
 
-	"codeberg.org/gruf/go-kv"
 	"github.com/superseriousbusiness/gotosocial/internal/ap"
+	"github.com/superseriousbusiness/gotosocial/internal/db"
+	"github.com/superseriousbusiness/gotosocial/internal/gtserror"
+	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
 	"github.com/superseriousbusiness/gotosocial/internal/log"
 	"github.com/superseriousbusiness/gotosocial/internal/messages"
 )
@@ -34,43 +37,130 @@ import (
 //
 // The library makes this call only after acquiring a lock first.
 func (f *federatingDB) Delete(ctx context.Context, id *url.URL) error {
-	l := log.WithContext(ctx).
-		WithFields(kv.Fields{
-			{"id", id},
-		}...)
-	l.Debug("entering Delete")
-
 	activityContext := getActivityContext(ctx)
 	if activityContext.internal {
 		return nil // Already processed.
 	}
 
-	requestingAcct := activityContext.requestingAcct
-	receivingAcct := activityContext.receivingAcct
-
-	// in a delete we only get the URI, we can't know if we have a status or a profile or something else,
-	// so we have to try a few different things...
-	if s, err := f.state.DB.GetStatusByURI(ctx, id.String()); err == nil && requestingAcct.ID == s.AccountID {
-		l.Debugf("deleting status: %s", s.ID)
-		f.state.Workers.EnqueueFediAPI(ctx, messages.FromFediAPI{
-			APObjectType:      ap.ObjectNote,
-			APActivityType:    ap.ActivityDelete,
-			GTSModel:          s,
-			ReceivingAccount:  receivingAcct,
-			RequestingAccount: requestingAcct,
+	// Extract receiving / requesting accounts.
+	requesting := activityContext.requestingAcct
+	receiving := activityContext.receivingAcct
+
+	// Serialize deleted ID URI.
+	// (may be status OR account)
+	uriStr := id.String()
+
+	var (
+		ok  bool
+		err error
+	)
+
+	// Try delete as an account URI.
+	ok, err = f.deleteAccount(ctx,
+		requesting,
+		receiving,
+		uriStr,
+	)
+	if err != nil {
+		return err
+	} else if ok {
+		// success!
+		return nil
+	}
+
+	// Try delete as a status URI.
+	ok, err = f.deleteStatus(ctx,
+		requesting,
+		receiving,
+		uriStr,
+	)
+	if err != nil {
+		return err
+	} else if ok {
+		// success!
+		return nil
+	}
+
+	// Log at debug level, as lots of these could indicate federation
+	// issues between remote and this instance, or help with debugging.
+	log.Debugf(ctx, "received delete for unknown target: %s", uriStr)
+	return nil
+}
+
+func (f *federatingDB) deleteAccount(
+	ctx context.Context,
+	requesting *gtsmodel.Account,
+	receiving *gtsmodel.Account,
+	uri string, // target account
+) (
+	bool, // success?
+	error, // any error
+) {
+	account, err := f.state.DB.GetAccountByURI(ctx, uri)
+	if err != nil && !errors.Is(err, db.ErrNoEntries) {
+		return false, gtserror.Newf("error getting account: %w", err)
+	}
+
+	if account != nil {
+		// Ensure requesting account is
+		// only trying to delete itself.
+		if account.ID != requesting.ID {
+
+			// TODO: handled forwarded deletes,
+			// for now we silently drop this.
+			return true, nil
+		}
+
+		log.Debugf(ctx, "deleting account: %s", account.URI)
+		f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{
+			APObjectType:   ap.ObjectProfile,
+			APActivityType: ap.ActivityDelete,
+			GTSModel:       account,
+			Receiving:      receiving,
+			Requesting:     requesting,
 		})
+
+		return true, nil
+	}
+
+	return false, nil
+}
+
+func (f *federatingDB) deleteStatus(
+	ctx context.Context,
+	requesting *gtsmodel.Account,
+	receiving *gtsmodel.Account,
+	uri string, // target status
+) (
+	bool, // success?
+	error, // any error
+) {
+	status, err := f.state.DB.GetStatusByURI(ctx, uri)
+	if err != nil && !errors.Is(err, db.ErrNoEntries) {
+		return false, gtserror.Newf("error getting status: %w", err)
 	}
 
-	if a, err := f.state.DB.GetAccountByURI(ctx, id.String()); err == nil && requestingAcct.ID == a.ID {
-		l.Debugf("deleting account: %s", a.ID)
-		f.state.Workers.EnqueueFediAPI(ctx, messages.FromFediAPI{
-			APObjectType:      ap.ObjectProfile,
-			APActivityType:    ap.ActivityDelete,
-			GTSModel:          a,
-			ReceivingAccount:  receivingAcct,
-			RequestingAccount: requestingAcct,
+	if status != nil {
+		// Ensure requesting account is only
+		// trying to delete its own statuses.
+		if status.AccountID != requesting.ID {
+
+			// TODO: handled forwarded deletes,
+			// for now we silently drop this.
+			return true, nil
+		}
+
+		log.Debugf(ctx, "deleting status: %s", status.URI)
+		f.state.Workers.Federator.Queue.Push(&messages.FromFediAPI{
+			APObjectType:   ap.ObjectNote,
+			APActivityType: ap.ActivityDelete,
+			GTSModel:       status,
+			Receiving:      receiving,
+			Requesting:     requesting,
 		})
+
+		return true, nil
 	}
 
-	return nil
+	return false, nil
 }