replace async client API / federator msg processing with worker pools (#497)

* replace async client API / federator msg processing with worker pools
* appease our lord-and-saviour, the linter

internal/processing/account/account.go

@@ -33,6 +33,7 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/text"
 	"github.com/superseriousbusiness/gotosocial/internal/typeutils"
 	"github.com/superseriousbusiness/gotosocial/internal/visibility"
+	"github.com/superseriousbusiness/gotosocial/internal/worker"
 	"github.com/superseriousbusiness/oauth2/v4"
 )
 
@@ -81,28 +82,28 @@ type Processor interface {
 }
 
 type processor struct {
-	tc            typeutils.TypeConverter
-	mediaManager  media.Manager
-	fromClientAPI chan messages.FromClientAPI
-	oauthServer   oauth.Server
-	filter        visibility.Filter
-	formatter     text.Formatter
-	db            db.DB
-	federator     federation.Federator
-	parseMention  gtsmodel.ParseMentionFunc
+	tc           typeutils.TypeConverter
+	mediaManager media.Manager
+	clientWorker *worker.Worker[messages.FromClientAPI]
+	oauthServer  oauth.Server
+	filter       visibility.Filter
+	formatter    text.Formatter
+	db           db.DB
+	federator    federation.Federator
+	parseMention gtsmodel.ParseMentionFunc
 }
 
 // New returns a new account processor.
-func New(db db.DB, tc typeutils.TypeConverter, mediaManager media.Manager, oauthServer oauth.Server, fromClientAPI chan messages.FromClientAPI, federator federation.Federator, parseMention gtsmodel.ParseMentionFunc) Processor {
+func New(db db.DB, tc typeutils.TypeConverter, mediaManager media.Manager, oauthServer oauth.Server, clientWorker *worker.Worker[messages.FromClientAPI], federator federation.Federator, parseMention gtsmodel.ParseMentionFunc) Processor {
 	return &processor{
-		tc:            tc,
-		mediaManager:  mediaManager,
-		fromClientAPI: fromClientAPI,
-		oauthServer:   oauthServer,
-		filter:        visibility.NewFilter(db),
-		formatter:     text.NewFormatter(db),
-		db:            db,
-		federator:     federator,
-		parseMention:  parseMention,
+		tc:           tc,
+		mediaManager: mediaManager,
+		clientWorker: clientWorker,
+		oauthServer:  oauthServer,
+		filter:       visibility.NewFilter(db),
+		formatter:    text.NewFormatter(db),
+		db:           db,
+		federator:    federator,
+		parseMention: parseMention,
 	}
 }

internal/processing/account/account_test.go

@@ -19,6 +19,8 @@
 package account_test
 
 import (
+	"context"
+
 	"codeberg.org/gruf/go-store/kv"
 	"github.com/stretchr/testify/suite"
 	"github.com/superseriousbusiness/activity/pub"
@@ -33,6 +35,7 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/processing/account"
 	"github.com/superseriousbusiness/gotosocial/internal/transport"
 	"github.com/superseriousbusiness/gotosocial/internal/typeutils"
+	"github.com/superseriousbusiness/gotosocial/internal/worker"
 	"github.com/superseriousbusiness/gotosocial/testrig"
 )
 
@@ -78,6 +81,16 @@ func (suite *AccountStandardTestSuite) SetupTest() {
 	testrig.InitTestLog()
 	testrig.InitTestConfig()
 
+	fedWorker := worker.New[messages.FromFederator](-1, -1)
+	clientWorker := worker.New[messages.FromClientAPI](-1, -1)
+	clientWorker.SetProcessor(func(_ context.Context, msg messages.FromClientAPI) error {
+		suite.fromClientAPIChan <- msg
+		return nil
+	})
+
+	_ = fedWorker.Start()
+	_ = clientWorker.Start()
+
 	suite.db = testrig.NewTestDB()
 	suite.tc = testrig.NewTestTypeConverter(suite.db)
 	suite.storage = testrig.NewTestStorage()
@@ -85,11 +98,11 @@ func (suite *AccountStandardTestSuite) SetupTest() {
 	suite.oauthServer = testrig.NewTestOauthServer(suite.db)
 	suite.fromClientAPIChan = make(chan messages.FromClientAPI, 100)
 	suite.httpClient = testrig.NewMockHTTPClient(nil)
-	suite.transportController = testrig.NewTestTransportController(suite.httpClient, suite.db)
-	suite.federator = testrig.NewTestFederator(suite.db, suite.transportController, suite.storage, suite.mediaManager)
+	suite.transportController = testrig.NewTestTransportController(suite.httpClient, suite.db, fedWorker)
+	suite.federator = testrig.NewTestFederator(suite.db, suite.transportController, suite.storage, suite.mediaManager, fedWorker)
 	suite.sentEmails = make(map[string]string)
 	suite.emailSender = testrig.NewEmailSender("../../../web/template/", suite.sentEmails)
-	suite.accountProcessor = account.New(suite.db, suite.tc, suite.mediaManager, suite.oauthServer, suite.fromClientAPIChan, suite.federator, processing.GetParseMentionFunc(suite.db, suite.federator))
+	suite.accountProcessor = account.New(suite.db, suite.tc, suite.mediaManager, suite.oauthServer, clientWorker, suite.federator, processing.GetParseMentionFunc(suite.db, suite.federator))
 	testrig.StandardDBSetup(suite.db, nil)
 	testrig.StandardStorageSetup(suite.storage, "../../../testrig/media")
 }

internal/processing/account/create.go

@@ -85,12 +85,12 @@ func (p *processor) Create(ctx context.Context, applicationToken oauth2.TokenInf
 
 	// there are side effects for creating a new account (sending confirmation emails etc)
 	// so pass a message to the processor so that it can do it asynchronously
-	p.fromClientAPI <- messages.FromClientAPI{
+	p.clientWorker.Queue(messages.FromClientAPI{
 		APObjectType:   ap.ObjectProfile,
 		APActivityType: ap.ActivityCreate,
 		GTSModel:       user.Account,
 		OriginAccount:  user.Account,
-	}
+	})
 
 	return &apimodel.Token{
 		AccessToken: accessToken.GetAccess(),

internal/processing/account/createblock.go

@@ -113,7 +113,7 @@ func (p *processor) BlockCreate(ctx context.Context, requestingAccount *gtsmodel
 
 	// follow request status changed so send the UNDO activity to the channel for async processing
 	if frChanged {
-		p.fromClientAPI <- messages.FromClientAPI{
+		p.clientWorker.Queue(messages.FromClientAPI{
 			APObjectType:   ap.ActivityFollow,
 			APActivityType: ap.ActivityUndo,
 			GTSModel: &gtsmodel.Follow{
@@ -123,12 +123,12 @@ func (p *processor) BlockCreate(ctx context.Context, requestingAccount *gtsmodel
 			},
 			OriginAccount: requestingAccount,
 			TargetAccount: targetAccount,
-		}
+		})
 	}
 
 	// follow status changed so send the UNDO activity to the channel for async processing
 	if fChanged {
-		p.fromClientAPI <- messages.FromClientAPI{
+		p.clientWorker.Queue(messages.FromClientAPI{
 			APObjectType:   ap.ActivityFollow,
 			APActivityType: ap.ActivityUndo,
 			GTSModel: &gtsmodel.Follow{
@@ -138,17 +138,17 @@ func (p *processor) BlockCreate(ctx context.Context, requestingAccount *gtsmodel
 			},
 			OriginAccount: requestingAccount,
 			TargetAccount: targetAccount,
-		}
+		})
 	}
 
 	// handle the rest of the block process asynchronously
-	p.fromClientAPI <- messages.FromClientAPI{
+	p.clientWorker.Queue(messages.FromClientAPI{
 		APObjectType:   ap.ActivityBlock,
 		APActivityType: ap.ActivityCreate,
 		GTSModel:       block,
 		OriginAccount:  requestingAccount,
 		TargetAccount:  targetAccount,
-	}
+	})
 
 	return p.RelationshipGet(ctx, requestingAccount, targetAccountID)
 }

internal/processing/account/createfollow.go

@@ -101,13 +101,13 @@ func (p *processor) FollowCreate(ctx context.Context, requestingAccount *gtsmode
 	}
 
 	// otherwise we leave the follow request as it is and we handle the rest of the process asynchronously
-	p.fromClientAPI <- messages.FromClientAPI{
+	p.clientWorker.Queue(messages.FromClientAPI{
 		APObjectType:   ap.ActivityFollow,
 		APActivityType: ap.ActivityCreate,
 		GTSModel:       fr,
 		OriginAccount:  requestingAccount,
 		TargetAccount:  targetAcct,
-	}
+	})
 
 	// return whatever relationship results from this
 	return p.RelationshipGet(ctx, requestingAccount, form.ID)

internal/processing/account/delete.go

@@ -159,13 +159,13 @@ selectStatusesLoop:
 			// pass the status delete through the client api channel for processing
 			s.Account = account
 			l.Debug("putting status in the client api channel")
-			p.fromClientAPI <- messages.FromClientAPI{
+			p.clientWorker.Queue(messages.FromClientAPI{
 				APObjectType:   ap.ObjectNote,
 				APActivityType: ap.ActivityDelete,
 				GTSModel:       s,
 				OriginAccount:  account,
 				TargetAccount:  account,
-			}
+			})
 
 			if err := p.db.DeleteByID(ctx, s.ID, s); err != nil {
 				if err != db.ErrNoEntries {
@@ -195,13 +195,13 @@ selectStatusesLoop:
 				}
 
 				l.Debug("putting boost undo in the client api channel")
-				p.fromClientAPI <- messages.FromClientAPI{
+				p.clientWorker.Queue(messages.FromClientAPI{
 					APObjectType:   ap.ActivityAnnounce,
 					APActivityType: ap.ActivityUndo,
 					GTSModel:       s,
 					OriginAccount:  b.Account,
 					TargetAccount:  account,
-				}
+				})
 
 				if err := p.db.DeleteByID(ctx, b.ID, b); err != nil {
 					if err != db.ErrNoEntries {
@@ -331,7 +331,7 @@ func (p *processor) DeleteLocal(ctx context.Context, account *gtsmodel.Account,
 	}
 
 	// put the delete in the processor queue to handle the rest of it asynchronously
-	p.fromClientAPI <- fromClientAPIMessage
+	p.clientWorker.Queue(fromClientAPIMessage)
 
 	return nil
 }

internal/processing/account/removeblock.go

@@ -54,13 +54,13 @@ func (p *processor) BlockRemove(ctx context.Context, requestingAccount *gtsmodel
 
 	// block status changed so send the UNDO activity to the channel for async processing
 	if blockChanged {
-		p.fromClientAPI <- messages.FromClientAPI{
+		p.clientWorker.Queue(messages.FromClientAPI{
 			APObjectType:   ap.ActivityBlock,
 			APActivityType: ap.ActivityUndo,
 			GTSModel:       block,
 			OriginAccount:  requestingAccount,
 			TargetAccount:  targetAccount,
-		}
+		})
 	}
 
 	// return whatever relationship results from all this

internal/processing/account/removefollow.go

@@ -80,7 +80,7 @@ func (p *processor) FollowRemove(ctx context.Context, requestingAccount *gtsmode
 
 	// follow request status changed so send the UNDO activity to the channel for async processing
 	if frChanged {
-		p.fromClientAPI <- messages.FromClientAPI{
+		p.clientWorker.Queue(messages.FromClientAPI{
 			APObjectType:   ap.ActivityFollow,
 			APActivityType: ap.ActivityUndo,
 			GTSModel: &gtsmodel.Follow{
@@ -90,12 +90,12 @@ func (p *processor) FollowRemove(ctx context.Context, requestingAccount *gtsmode
 			},
 			OriginAccount: requestingAccount,
 			TargetAccount: targetAcct,
-		}
+		})
 	}
 
 	// follow status changed so send the UNDO activity to the channel for async processing
 	if fChanged {
-		p.fromClientAPI <- messages.FromClientAPI{
+		p.clientWorker.Queue(messages.FromClientAPI{
 			APObjectType:   ap.ActivityFollow,
 			APActivityType: ap.ActivityUndo,
 			GTSModel: &gtsmodel.Follow{
@@ -105,7 +105,7 @@ func (p *processor) FollowRemove(ctx context.Context, requestingAccount *gtsmode
 			},
 			OriginAccount: requestingAccount,
 			TargetAccount: targetAcct,
-		}
+		})
 	}
 
 	// return whatever relationship results from all this

internal/processing/account/update.go

@@ -117,12 +117,12 @@ func (p *processor) Update(ctx context.Context, account *gtsmodel.Account, form
 		return nil, fmt.Errorf("could not update account %s: %s", account.ID, err)
 	}
 
-	p.fromClientAPI <- messages.FromClientAPI{
+	p.clientWorker.Queue(messages.FromClientAPI{
 		APObjectType:   ap.ObjectProfile,
 		APActivityType: ap.ActivityUpdate,
 		GTSModel:       updatedAccount,
 		OriginAccount:  updatedAccount,
-	}
+	})
 
 	acctSensitive, err := p.tc.AccountToAPIAccountSensitive(ctx, updatedAccount)
 	if err != nil {

internal/processing/admin/accountaction.go

@@ -34,12 +34,12 @@ func (p *processor) AccountAction(ctx context.Context, account *gtsmodel.Account
 	case string(gtsmodel.AdminActionSuspend):
 		adminAction.Type = gtsmodel.AdminActionSuspend
 		// pass the account delete through the client api channel for processing
-		p.fromClientAPI <- messages.FromClientAPI{
+		p.clientWorker.Queue(messages.FromClientAPI{
 			APObjectType:   ap.ActorPerson,
 			APActivityType: ap.ActivityDelete,
 			OriginAccount:  account,
 			TargetAccount:  targetAccount,
-		}
+		})
 	default:
 		return gtserror.NewErrorBadRequest(fmt.Errorf("admin action type %s is not supported for this endpoint", form.Type))
 	}

internal/processing/admin/admin.go

@@ -29,6 +29,7 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/media"
 	"github.com/superseriousbusiness/gotosocial/internal/messages"
 	"github.com/superseriousbusiness/gotosocial/internal/typeutils"
+	"github.com/superseriousbusiness/gotosocial/internal/worker"
 )
 
 // Processor wraps a bunch of functions for processing admin actions.
@@ -43,18 +44,18 @@ type Processor interface {
 }
 
 type processor struct {
-	tc            typeutils.TypeConverter
-	mediaManager  media.Manager
-	fromClientAPI chan messages.FromClientAPI
-	db            db.DB
+	tc           typeutils.TypeConverter
+	mediaManager media.Manager
+	clientWorker *worker.Worker[messages.FromClientAPI]
+	db           db.DB
 }
 
 // New returns a new admin processor.
-func New(db db.DB, tc typeutils.TypeConverter, mediaManager media.Manager, fromClientAPI chan messages.FromClientAPI) Processor {
+func New(db db.DB, tc typeutils.TypeConverter, mediaManager media.Manager, clientWorker *worker.Worker[messages.FromClientAPI]) Processor {
 	return &processor{
-		tc:            tc,
-		mediaManager:  mediaManager,
-		fromClientAPI: fromClientAPI,
-		db:            db,
+		tc:           tc,
+		mediaManager: mediaManager,
+		clientWorker: clientWorker,
+		db:           db,
 	}
 }

internal/processing/admin/createdomainblock.go

@@ -141,13 +141,13 @@ selectAccountsLoop:
 			l.Debugf("putting delete for account %s in the clientAPI channel", a.Username)
 
 			// pass the account delete through the client api channel for processing
-			p.fromClientAPI <- messages.FromClientAPI{
+			p.clientWorker.Queue(messages.FromClientAPI{
 				APObjectType:   ap.ActorPerson,
 				APActivityType: ap.ActivityDelete,
 				GTSModel:       block,
 				OriginAccount:  account,
 				TargetAccount:  a,
-			}
+			})
 
 			// if this is the last account in the slice, set the maxID appropriately for the next query
 			if i == len(accounts)-1 {

internal/processing/federation/federation.go

@@ -27,7 +27,6 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/db"
 	"github.com/superseriousbusiness/gotosocial/internal/federation"
 	"github.com/superseriousbusiness/gotosocial/internal/gtserror"
-	"github.com/superseriousbusiness/gotosocial/internal/messages"
 	"github.com/superseriousbusiness/gotosocial/internal/typeutils"
 	"github.com/superseriousbusiness/gotosocial/internal/visibility"
 )
@@ -81,20 +80,18 @@ type Processor interface {
 }
 
 type processor struct {
-	db            db.DB
-	federator     federation.Federator
-	tc            typeutils.TypeConverter
-	filter        visibility.Filter
-	fromFederator chan messages.FromFederator
+	db        db.DB
+	federator federation.Federator
+	tc        typeutils.TypeConverter
+	filter    visibility.Filter
 }
 
 // New returns a new federation processor.
-func New(db db.DB, tc typeutils.TypeConverter, federator federation.Federator, fromFederator chan messages.FromFederator) Processor {
+func New(db db.DB, tc typeutils.TypeConverter, federator federation.Federator) Processor {
 	return &processor{
-		db:            db,
-		federator:     federator,
-		tc:            tc,
-		filter:        visibility.NewFilter(db),
-		fromFederator: fromFederator,
+		db:        db,
+		federator: federator,
+		tc:        tc,
+		filter:    visibility.NewFilter(db),
 	}
 }

internal/processing/federation/postinbox.go

@@ -21,12 +21,8 @@ package federation
 import (
 	"context"
 	"net/http"
-
-	"github.com/superseriousbusiness/gotosocial/internal/ap"
 )
 
 func (p *processor) PostInbox(ctx context.Context, w http.ResponseWriter, r *http.Request) (bool, error) {
-	// pass the fromFederator channel through to postInbox, since it'll be needed later
-	contextWithChannel := context.WithValue(ctx, ap.ContextFromFederatorChan, p.fromFederator)
-	return p.federator.FederatingActor().PostInbox(contextWithChannel, w, r)
+	return p.federator.FederatingActor().PostInbox(ctx, w, r)
 }

internal/processing/followrequest.go

@@ -78,13 +78,13 @@ func (p *processor) FollowRequestAccept(ctx context.Context, auth *oauth.Auth, a
 		follow.TargetAccount = followTargetAccount
 	}
 
-	p.fromClientAPI <- messages.FromClientAPI{
+	p.clientWorker.Queue(messages.FromClientAPI{
 		APObjectType:   ap.ActivityFollow,
 		APActivityType: ap.ActivityAccept,
 		GTSModel:       follow,
 		OriginAccount:  follow.Account,
 		TargetAccount:  follow.TargetAccount,
-	}
+	})
 
 	gtsR, err := p.db.GetRelationship(ctx, auth.Account.ID, accountID)
 	if err != nil {
@@ -121,13 +121,13 @@ func (p *processor) FollowRequestReject(ctx context.Context, auth *oauth.Auth, a
 		followRequest.TargetAccount = a
 	}
 
-	p.fromClientAPI <- messages.FromClientAPI{
+	p.clientWorker.Queue(messages.FromClientAPI{
 		APObjectType:   ap.ActivityFollow,
 		APActivityType: ap.ActivityReject,
 		GTSModel:       followRequest,
 		OriginAccount:  followRequest.Account,
 		TargetAccount:  followRequest.TargetAccount,
-	}
+	})
 
 	gtsR, err := p.db.GetRelationship(ctx, auth.Account.ID, accountID)
 	if err != nil {

internal/processing/media/media_test.go

@@ -29,9 +29,11 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/db"
 	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
 	"github.com/superseriousbusiness/gotosocial/internal/media"
+	"github.com/superseriousbusiness/gotosocial/internal/messages"
 	mediaprocessing "github.com/superseriousbusiness/gotosocial/internal/processing/media"
 	"github.com/superseriousbusiness/gotosocial/internal/transport"
 	"github.com/superseriousbusiness/gotosocial/internal/typeutils"
+	"github.com/superseriousbusiness/gotosocial/internal/worker"
 	"github.com/superseriousbusiness/gotosocial/testrig"
 )
 
@@ -120,6 +122,7 @@ func (suite *MediaStandardTestSuite) mockTransportController() transport.Control
 
 		return response, nil
 	}
+	fedWorker := worker.New[messages.FromFederator](-1, -1)
 	mockClient := testrig.NewMockHTTPClient(do)
-	return testrig.NewTestTransportController(mockClient, suite.db)
+	return testrig.NewTestTransportController(mockClient, suite.db, fedWorker)
 }

internal/processing/processor.go

@@ -24,7 +24,6 @@ import (
 	"net/url"
 
 	"codeberg.org/gruf/go-store/kv"
-	"github.com/sirupsen/logrus"
 	apimodel "github.com/superseriousbusiness/gotosocial/internal/api/model"
 	"github.com/superseriousbusiness/gotosocial/internal/db"
 	"github.com/superseriousbusiness/gotosocial/internal/email"
@@ -45,6 +44,7 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/timeline"
 	"github.com/superseriousbusiness/gotosocial/internal/typeutils"
 	"github.com/superseriousbusiness/gotosocial/internal/visibility"
+	"github.com/superseriousbusiness/gotosocial/internal/worker"
 )
 
 // Processor should be passed to api modules (see internal/apimodule/...). It is used for
@@ -55,7 +55,7 @@ import (
 // for clean distribution of messages without slowing down the client API and harming the user experience.
 type Processor interface {
 	// Start starts the Processor, reading from its channels and passing messages back and forth.
-	Start(ctx context.Context) error
+	Start() error
 	// Stop stops the processor cleanly, finishing handling any remaining messages before closing down.
 	Stop() error
 	// ProcessFromClientAPI processes one message coming from the clientAPI channel, and triggers appropriate side effects.
@@ -235,10 +235,10 @@ type Processor interface {
 
 // processor just implements the Processor interface
 type processor struct {
-	fromClientAPI   chan messages.FromClientAPI
-	fromFederator   chan messages.FromFederator
+	clientWorker *worker.Worker[messages.FromClientAPI]
+	fedWorker    *worker.Worker[messages.FromFederator]
+
 	federator       federation.Federator
-	stop            chan interface{}
 	tc              typeutils.TypeConverter
 	oauthServer     oauth.Server
 	mediaManager    media.Manager
@@ -268,26 +268,26 @@ func NewProcessor(
 	mediaManager media.Manager,
 	storage *kv.KVStore,
 	db db.DB,
-	emailSender email.Sender) Processor {
-
-	fromClientAPI := make(chan messages.FromClientAPI, 1000)
-	fromFederator := make(chan messages.FromFederator, 1000)
+	emailSender email.Sender,
+	clientWorker *worker.Worker[messages.FromClientAPI],
+	fedWorker *worker.Worker[messages.FromFederator],
+) Processor {
 	parseMentionFunc := GetParseMentionFunc(db, federator)
 
-	statusProcessor := status.New(db, tc, fromClientAPI, parseMentionFunc)
+	statusProcessor := status.New(db, tc, clientWorker, parseMentionFunc)
 	streamingProcessor := streaming.New(db, oauthServer)
-	accountProcessor := account.New(db, tc, mediaManager, oauthServer, fromClientAPI, federator, parseMentionFunc)
-	adminProcessor := admin.New(db, tc, mediaManager, fromClientAPI)
+	accountProcessor := account.New(db, tc, mediaManager, oauthServer, clientWorker, federator, parseMentionFunc)
+	adminProcessor := admin.New(db, tc, mediaManager, clientWorker)
 	mediaProcessor := mediaProcessor.New(db, tc, mediaManager, federator.TransportController(), storage)
 	userProcessor := user.New(db, emailSender)
-	federationProcessor := federationProcessor.New(db, tc, federator, fromFederator)
+	federationProcessor := federationProcessor.New(db, tc, federator)
 	filter := visibility.NewFilter(db)
 
 	return &processor{
-		fromClientAPI:   fromClientAPI,
-		fromFederator:   fromFederator,
+		clientWorker: clientWorker,
+		fedWorker:    fedWorker,
+
 		federator:       federator,
-		stop:            make(chan interface{}),
 		tc:              tc,
 		oauthServer:     oauthServer,
 		mediaManager:    mediaManager,
@@ -307,36 +307,29 @@ func NewProcessor(
 }
 
 // Start starts the Processor, reading from its channels and passing messages back and forth.
-func (p *processor) Start(ctx context.Context) error {
-	go func() {
-	DistLoop:
-		for {
-			select {
-			case clientMsg := <-p.fromClientAPI:
-				logrus.Tracef("received message FROM client API: %+v", clientMsg)
-				go func() {
-					if err := p.ProcessFromClientAPI(ctx, clientMsg); err != nil {
-						logrus.Error(err)
-					}
-				}()
-			case federatorMsg := <-p.fromFederator:
-				logrus.Tracef("received message FROM federator: %+v", federatorMsg)
-				go func() {
-					if err := p.ProcessFromFederator(ctx, federatorMsg); err != nil {
-						logrus.Error(err)
-					}
-				}()
-			case <-p.stop:
-				break DistLoop
-			}
-		}
-	}()
+func (p *processor) Start() error {
+	// Setup and start the client API worker pool
+	p.clientWorker.SetProcessor(p.ProcessFromClientAPI)
+	if err := p.clientWorker.Start(); err != nil {
+		return err
+	}
+
+	// Setup and start the federator worker pool
+	p.fedWorker.SetProcessor(p.ProcessFromFederator)
+	if err := p.fedWorker.Start(); err != nil {
+		return err
+	}
+
 	return nil
 }
 
 // Stop stops the processor cleanly, finishing handling any remaining messages before closing down.
-// TODO: empty message buffer properly before stopping otherwise we'll lose federating messages.
 func (p *processor) Stop() error {
-	close(p.stop)
+	if err := p.clientWorker.Stop(); err != nil {
+		return err
+	}
+	if err := p.fedWorker.Stop(); err != nil {
+		return err
+	}
 	return nil
 }

internal/processing/processor_test.go

@@ -34,11 +34,13 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/federation"
 	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
 	"github.com/superseriousbusiness/gotosocial/internal/media"
+	"github.com/superseriousbusiness/gotosocial/internal/messages"
 	"github.com/superseriousbusiness/gotosocial/internal/oauth"
 	"github.com/superseriousbusiness/gotosocial/internal/processing"
 	"github.com/superseriousbusiness/gotosocial/internal/timeline"
 	"github.com/superseriousbusiness/gotosocial/internal/transport"
 	"github.com/superseriousbusiness/gotosocial/internal/typeutils"
+	"github.com/superseriousbusiness/gotosocial/internal/worker"
 	"github.com/superseriousbusiness/gotosocial/testrig"
 )
 
@@ -215,17 +217,20 @@ func (suite *ProcessingStandardTestSuite) SetupTest() {
 		}, nil
 	})
 
-	suite.transportController = testrig.NewTestTransportController(httpClient, suite.db)
+	clientWorker := worker.New[messages.FromClientAPI](-1, -1)
+	fedWorker := worker.New[messages.FromFederator](-1, -1)
+
+	suite.transportController = testrig.NewTestTransportController(httpClient, suite.db, fedWorker)
 	suite.mediaManager = testrig.NewTestMediaManager(suite.db, suite.storage)
-	suite.federator = testrig.NewTestFederator(suite.db, suite.transportController, suite.storage, suite.mediaManager)
+	suite.federator = testrig.NewTestFederator(suite.db, suite.transportController, suite.storage, suite.mediaManager, fedWorker)
 	suite.oauthServer = testrig.NewTestOauthServer(suite.db)
 	suite.emailSender = testrig.NewEmailSender("../../web/template/", nil)
 
-	suite.processor = processing.NewProcessor(suite.typeconverter, suite.federator, suite.oauthServer, suite.mediaManager, suite.storage, suite.db, suite.emailSender)
+	suite.processor = processing.NewProcessor(suite.typeconverter, suite.federator, suite.oauthServer, suite.mediaManager, suite.storage, suite.db, suite.emailSender, clientWorker, fedWorker)
 
 	testrig.StandardDBSetup(suite.db, suite.testAccounts)
 	testrig.StandardStorageSetup(suite.storage, "../../testrig/media")
-	if err := suite.processor.Start(context.Background()); err != nil {
+	if err := suite.processor.Start(); err != nil {
 		panic(err)
 	}
 }

internal/processing/status/boost.go

@@ -65,13 +65,13 @@ func (p *processor) Boost(ctx context.Context, requestingAccount *gtsmodel.Accou
 	}
 
 	// send it back to the processor for async processing
-	p.fromClientAPI <- messages.FromClientAPI{
+	p.clientWorker.Queue(messages.FromClientAPI{
 		APObjectType:   ap.ActivityAnnounce,
 		APActivityType: ap.ActivityCreate,
 		GTSModel:       boostWrapperStatus,
 		OriginAccount:  requestingAccount,
 		TargetAccount:  targetStatus.Account,
-	}
+	})
 
 	// return the frontend representation of the new status to the submitter
 	apiStatus, err := p.tc.StatusToAPIStatus(ctx, boostWrapperStatus, requestingAccount)

internal/processing/status/create.go

@@ -97,12 +97,12 @@ func (p *processor) Create(ctx context.Context, account *gtsmodel.Account, appli
 	}
 
 	// send it back to the processor for async processing
-	p.fromClientAPI <- messages.FromClientAPI{
+	p.clientWorker.Queue(messages.FromClientAPI{
 		APObjectType:   ap.ObjectNote,
 		APActivityType: ap.ActivityCreate,
 		GTSModel:       newStatus,
 		OriginAccount:  account,
-	}
+	})
 
 	// return the frontend representation of the new status to the submitter
 	apiStatus, err := p.tc.StatusToAPIStatus(ctx, newStatus, account)

internal/processing/status/delete.go

@@ -53,13 +53,13 @@ func (p *processor) Delete(ctx context.Context, requestingAccount *gtsmodel.Acco
 	}
 
 	// send it back to the processor for async processing
-	p.fromClientAPI <- messages.FromClientAPI{
+	p.clientWorker.Queue(messages.FromClientAPI{
 		APObjectType:   ap.ObjectNote,
 		APActivityType: ap.ActivityDelete,
 		GTSModel:       targetStatus,
 		OriginAccount:  requestingAccount,
 		TargetAccount:  requestingAccount,
-	}
+	})
 
 	return apiStatus, nil
 }

internal/processing/status/fave.go

@@ -84,13 +84,13 @@ func (p *processor) Fave(ctx context.Context, requestingAccount *gtsmodel.Accoun
 		}
 
 		// send it back to the processor for async processing
-		p.fromClientAPI <- messages.FromClientAPI{
+		p.clientWorker.Queue(messages.FromClientAPI{
 			APObjectType:   ap.ActivityLike,
 			APActivityType: ap.ActivityCreate,
 			GTSModel:       gtsFave,
 			OriginAccount:  requestingAccount,
 			TargetAccount:  targetStatus.Account,
-		}
+		})
 	}
 
 	// return the apidon representation of the target status

internal/processing/status/status.go

@@ -29,6 +29,7 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/text"
 	"github.com/superseriousbusiness/gotosocial/internal/typeutils"
 	"github.com/superseriousbusiness/gotosocial/internal/visibility"
+	"github.com/superseriousbusiness/gotosocial/internal/worker"
 )
 
 // Processor wraps a bunch of functions for processing statuses.
@@ -69,22 +70,22 @@ type Processor interface {
 }
 
 type processor struct {
-	tc            typeutils.TypeConverter
-	db            db.DB
-	filter        visibility.Filter
-	formatter     text.Formatter
-	fromClientAPI chan messages.FromClientAPI
-	parseMention  gtsmodel.ParseMentionFunc
+	tc           typeutils.TypeConverter
+	db           db.DB
+	filter       visibility.Filter
+	formatter    text.Formatter
+	clientWorker *worker.Worker[messages.FromClientAPI]
+	parseMention gtsmodel.ParseMentionFunc
 }
 
 // New returns a new status processor.
-func New(db db.DB, tc typeutils.TypeConverter, fromClientAPI chan messages.FromClientAPI, parseMention gtsmodel.ParseMentionFunc) Processor {
+func New(db db.DB, tc typeutils.TypeConverter, clientWorker *worker.Worker[messages.FromClientAPI], parseMention gtsmodel.ParseMentionFunc) Processor {
 	return &processor{
-		tc:            tc,
-		db:            db,
-		filter:        visibility.NewFilter(db),
-		formatter:     text.NewFormatter(db),
-		fromClientAPI: fromClientAPI,
-		parseMention:  parseMention,
+		tc:           tc,
+		db:           db,
+		filter:       visibility.NewFilter(db),
+		formatter:    text.NewFormatter(db),
+		clientWorker: clientWorker,
+		parseMention: parseMention,
 	}
 }

internal/processing/status/status_test.go

@@ -30,18 +30,19 @@ import (
 	"github.com/superseriousbusiness/gotosocial/internal/processing/status"
 	"github.com/superseriousbusiness/gotosocial/internal/transport"
 	"github.com/superseriousbusiness/gotosocial/internal/typeutils"
+	"github.com/superseriousbusiness/gotosocial/internal/worker"
 	"github.com/superseriousbusiness/gotosocial/testrig"
 )
 
 type StatusStandardTestSuite struct {
 	suite.Suite
-	db                db.DB
-	typeConverter     typeutils.TypeConverter
-	tc                transport.Controller
-	storage           *kv.KVStore
-	mediaManager      media.Manager
-	federator         federation.Federator
-	fromClientAPIChan chan messages.FromClientAPI
+	db            db.DB
+	typeConverter typeutils.TypeConverter
+	tc            transport.Controller
+	storage       *kv.KVStore
+	mediaManager  media.Manager
+	federator     federation.Federator
+	clientWorker  *worker.Worker[messages.FromClientAPI]
 
 	// standard suite models
 	testTokens       map[string]*gtsmodel.Token
@@ -74,14 +75,16 @@ func (suite *StatusStandardTestSuite) SetupTest() {
 	testrig.InitTestConfig()
 	testrig.InitTestLog()
 
+	fedWorker := worker.New[messages.FromFederator](-1, -1)
+
 	suite.db = testrig.NewTestDB()
 	suite.typeConverter = testrig.NewTestTypeConverter(suite.db)
-	suite.fromClientAPIChan = make(chan messages.FromClientAPI, 100)
-	suite.tc = testrig.NewTestTransportController(testrig.NewMockHTTPClient(nil), suite.db)
+	suite.clientWorker = worker.New[messages.FromClientAPI](-1, -1)
+	suite.tc = testrig.NewTestTransportController(testrig.NewMockHTTPClient(nil), suite.db, fedWorker)
 	suite.storage = testrig.NewTestStorage()
 	suite.mediaManager = testrig.NewTestMediaManager(suite.db, suite.storage)
-	suite.federator = testrig.NewTestFederator(suite.db, suite.tc, suite.storage, suite.mediaManager)
-	suite.status = status.New(suite.db, suite.typeConverter, suite.fromClientAPIChan, processing.GetParseMentionFunc(suite.db, suite.federator))
+	suite.federator = testrig.NewTestFederator(suite.db, suite.tc, suite.storage, suite.mediaManager, fedWorker)
+	suite.status = status.New(suite.db, suite.typeConverter, suite.clientWorker, processing.GetParseMentionFunc(suite.db, suite.federator))
 
 	testrig.StandardDBSetup(suite.db, suite.testAccounts)
 	testrig.StandardStorageSetup(suite.storage, "../../../testrig/media")

internal/processing/status/unboost.go

@@ -91,13 +91,13 @@ func (p *processor) Unboost(ctx context.Context, requestingAccount *gtsmodel.Acc
 		gtsBoost.BoostOf.Account = targetStatus.Account
 
 		// send it back to the processor for async processing
-		p.fromClientAPI <- messages.FromClientAPI{
+		p.clientWorker.Queue(messages.FromClientAPI{
 			APObjectType:   ap.ActivityAnnounce,
 			APActivityType: ap.ActivityUndo,
 			GTSModel:       gtsBoost,
 			OriginAccount:  requestingAccount,
 			TargetAccount:  targetStatus.Account,
-		}
+		})
 	}
 
 	apiStatus, err := p.tc.StatusToAPIStatus(ctx, targetStatus, requestingAccount)

internal/processing/status/unfave.go

@@ -73,13 +73,13 @@ func (p *processor) Unfave(ctx context.Context, requestingAccount *gtsmodel.Acco
 		}
 
 		// send it back to the processor for async processing
-		p.fromClientAPI <- messages.FromClientAPI{
+		p.clientWorker.Queue(messages.FromClientAPI{
 			APObjectType:   ap.ActivityLike,
 			APActivityType: ap.ActivityUndo,
 			GTSModel:       gtsFave,
 			OriginAccount:  requestingAccount,
 			TargetAccount:  targetStatus.Account,
-		}
+		})
 	}
 
 	apiStatus, err := p.tc.StatusToAPIStatus(ctx, targetStatus, requestingAccount)