[chore] move client/federator workerpools to Workers{} (#1575)

* replace concurrency worker pools with base models in State.Workers, update code and tests accordingly

* improve code comment

* change back testrig default log level

* un-comment-out TestAnnounceTwice() and fix

---------

Signed-off-by: kim <grufwub@gmail.com>
Reviewed-by: tobi

20 files changed, 127 insertions, 123 deletions

diff --git a/internal/federation/dereferencing/dereferencer_test.go b/internal/federation/dereferencing/dereferencer_test.go
index daca8b7de..f5b59b0ed 100644
--- a/internal/federation/dereferencing/dereferencer_test.go
+++ b/internal/federation/dereferencing/dereferencer_test.go
@@ -21,11 +21,10 @@ package dereferencing_test
 import (
 	"github.com/stretchr/testify/suite"
 	"github.com/superseriousbusiness/activity/streams/vocab"
-	"github.com/superseriousbusiness/gotosocial/internal/concurrency"
 	"github.com/superseriousbusiness/gotosocial/internal/db"
 	"github.com/superseriousbusiness/gotosocial/internal/federation/dereferencing"
 	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
-	"github.com/superseriousbusiness/gotosocial/internal/messages"
+	"github.com/superseriousbusiness/gotosocial/internal/state"
 	"github.com/superseriousbusiness/gotosocial/internal/storage"
 	"github.com/superseriousbusiness/gotosocial/testrig"
 )
@@ -34,6 +33,7 @@ type DereferencerStandardTestSuite struct {
 	suite.Suite
 	db      db.DB
 	storage *storage.Driver
+	state   state.State
 
 	testRemoteStatuses    map[string]vocab.ActivityStreamsNote
 	testRemotePeople      map[string]vocab.ActivityStreamsPerson
@@ -58,12 +58,19 @@ func (suite *DereferencerStandardTestSuite) SetupTest() {
 	suite.testRemoteAttachments = testrig.NewTestFediAttachments("../../../testrig/media")
 	suite.testEmojis = testrig.NewTestEmojis()
 
-	suite.db = testrig.NewTestDB()
+	suite.state.Caches.Init()
+	testrig.StartWorkers(&suite.state)
+
+	suite.db = testrig.NewTestDB(&suite.state)
 	suite.storage = testrig.NewInMemoryStorage()
-	suite.dereferencer = dereferencing.NewDereferencer(suite.db, testrig.NewTestTypeConverter(suite.db), testrig.NewTestTransportController(testrig.NewMockHTTPClient(nil, "../../../testrig/media"), suite.db, concurrency.NewWorkerPool[messages.FromFederator](-1, -1)), testrig.NewTestMediaManager(suite.db, suite.storage))
+	suite.state.DB = suite.db
+	suite.state.Storage = suite.storage
+	media := testrig.NewTestMediaManager(&suite.state)
+	suite.dereferencer = dereferencing.NewDereferencer(suite.db, testrig.NewTestTypeConverter(suite.db), testrig.NewTestTransportController(&suite.state, testrig.NewMockHTTPClient(nil, "../../../testrig/media")), media)
 	testrig.StandardDBSetup(suite.db, nil)
 }
 
 func (suite *DereferencerStandardTestSuite) TearDownTest() {
 	testrig.StandardDBTeardown(suite.db)
+	testrig.StopWorkers(&suite.state)
 }
diff --git a/internal/federation/federatingactor_test.go b/internal/federation/federatingactor_test.go
index 0d1d8e37f..f63ecd827 100644
--- a/internal/federation/federatingactor_test.go
+++ b/internal/federation/federatingactor_test.go
@@ -27,10 +27,8 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/suite"
-	"github.com/superseriousbusiness/gotosocial/internal/concurrency"
 	"github.com/superseriousbusiness/gotosocial/internal/federation"
 	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
-	"github.com/superseriousbusiness/gotosocial/internal/messages"
 	"github.com/superseriousbusiness/gotosocial/testrig"
 )
 
@@ -56,14 +54,12 @@ func (suite *FederatingActorTestSuite) TestSendNoRemoteFollowers() {
 	)
 	testActivity := testrig.WrapAPNoteInCreate(testrig.URLMustParse("http://localhost:8080/whatever_some_create"), testrig.URLMustParse(testAccount.URI), time.Now(), testNote)
 
-	fedWorker := concurrency.NewWorkerPool[messages.FromFederator](-1, -1)
-
 	// setup transport controller with a no-op client so we don't make external calls
 	httpClient := testrig.NewMockHTTPClient(nil, "../../testrig/media")
-	tc := testrig.NewTestTransportController(httpClient, suite.db, fedWorker)
+	tc := testrig.NewTestTransportController(&suite.state, httpClient)
 
 	// setup module being tested
-	federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(suite.db, fedWorker), tc, suite.tc, testrig.NewTestMediaManager(suite.db, suite.storage))
+	federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(&suite.state), tc, suite.tc, testrig.NewTestMediaManager(&suite.state))
 
 	activity, err := federator.FederatingActor().Send(ctx, testrig.URLMustParse(testAccount.OutboxURI), testActivity)
 	suite.NoError(err)
@@ -105,12 +101,10 @@ func (suite *FederatingActorTestSuite) TestSendRemoteFollower() {
 	)
 	testActivity := testrig.WrapAPNoteInCreate(testrig.URLMustParse("http://localhost:8080/whatever_some_create"), testrig.URLMustParse(testAccount.URI), testrig.TimeMustParse("2022-06-02T12:22:21+02:00"), testNote)
 
-	fedWorker := concurrency.NewWorkerPool[messages.FromFederator](-1, -1)
-
 	httpClient := testrig.NewMockHTTPClient(nil, "../../testrig/media")
-	tc := testrig.NewTestTransportController(httpClient, suite.db, fedWorker)
+	tc := testrig.NewTestTransportController(&suite.state, httpClient)
 	// setup module being tested
-	federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(suite.db, fedWorker), tc, suite.tc, testrig.NewTestMediaManager(suite.db, suite.storage))
+	federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(&suite.state), tc, suite.tc, testrig.NewTestMediaManager(&suite.state))
 
 	activity, err := federator.FederatingActor().Send(ctx, testrig.URLMustParse(testAccount.OutboxURI), testActivity)
 	suite.NoError(err)
diff --git a/internal/federation/federatingdb/accept.go b/internal/federation/federatingdb/accept.go
index d3e227a10..184d2b09d 100644
--- a/internal/federation/federatingdb/accept.go
+++ b/internal/federation/federatingdb/accept.go
@@ -65,7 +65,7 @@ func (f *federatingDB) Accept(ctx context.Context, accept vocab.ActivityStreamsA
 			if uris.IsFollowPath(acceptedObjectIRI) {
 				// ACCEPT FOLLOW
 				gtsFollowRequest := &gtsmodel.FollowRequest{}
-				if err := f.db.GetWhere(ctx, []db.Where{{Key: "uri", Value: acceptedObjectIRI.String()}}, gtsFollowRequest); err != nil {
+				if err := f.state.DB.GetWhere(ctx, []db.Where{{Key: "uri", Value: acceptedObjectIRI.String()}}, gtsFollowRequest); err != nil {
 					return fmt.Errorf("ACCEPT: couldn't get follow request with id %s from the database: %s", acceptedObjectIRI.String(), err)
 				}
 
@@ -73,12 +73,12 @@ func (f *federatingDB) Accept(ctx context.Context, accept vocab.ActivityStreamsA
 				if gtsFollowRequest.AccountID != receivingAccount.ID {
 					return errors.New("ACCEPT: follow object account and inbox account were not the same")
 				}
-				follow, err := f.db.AcceptFollowRequest(ctx, gtsFollowRequest.AccountID, gtsFollowRequest.TargetAccountID)
+				follow, err := f.state.DB.AcceptFollowRequest(ctx, gtsFollowRequest.AccountID, gtsFollowRequest.TargetAccountID)
 				if err != nil {
 					return err
 				}
 
-				f.fedWorker.Queue(messages.FromFederator{
+				f.state.Workers.EnqueueFederator(ctx, messages.FromFederator{
 					APObjectType:     ap.ActivityFollow,
 					APActivityType:   ap.ActivityAccept,
 					GTSModel:         follow,
@@ -108,12 +108,12 @@ func (f *federatingDB) Accept(ctx context.Context, accept vocab.ActivityStreamsA
 			if gtsFollow.AccountID != receivingAccount.ID {
 				return errors.New("ACCEPT: follow object account and inbox account were not the same")
 			}
-			follow, err := f.db.AcceptFollowRequest(ctx, gtsFollow.AccountID, gtsFollow.TargetAccountID)
+			follow, err := f.state.DB.AcceptFollowRequest(ctx, gtsFollow.AccountID, gtsFollow.TargetAccountID)
 			if err != nil {
 				return err
 			}
 
-			f.fedWorker.Queue(messages.FromFederator{
+			f.state.Workers.EnqueueFederator(ctx, messages.FromFederator{
 				APObjectType:     ap.ActivityFollow,
 				APActivityType:   ap.ActivityAccept,
 				GTSModel:         follow,
diff --git a/internal/federation/federatingdb/announce.go b/internal/federation/federatingdb/announce.go
index f4d145148..552a95ba9 100644
--- a/internal/federation/federatingdb/announce.go
+++ b/internal/federation/federatingdb/announce.go
@@ -59,7 +59,7 @@ func (f *federatingDB) Announce(ctx context.Context, announce vocab.ActivityStre
 	}
 
 	// it's a new announce so pass it back to the processor async for dereferencing etc
-	f.fedWorker.Queue(messages.FromFederator{
+	f.state.Workers.EnqueueFederator(ctx, messages.FromFederator{
 		APObjectType:     ap.ActivityAnnounce,
 		APActivityType:   ap.ActivityCreate,
 		GTSModel:         boost,
diff --git a/internal/federation/federatingdb/announce_test.go b/internal/federation/federatingdb/announce_test.go
index 6c0d969f4..d9158f383 100644
--- a/internal/federation/federatingdb/announce_test.go
+++ b/internal/federation/federatingdb/announce_test.go
@@ -25,6 +25,7 @@ import (
 	"github.com/superseriousbusiness/activity/streams/vocab"
 	"github.com/superseriousbusiness/gotosocial/internal/ap"
 	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
+	"github.com/superseriousbusiness/gotosocial/internal/id"
 )
 
 type AnnounceTestSuite struct {
@@ -74,6 +75,13 @@ func (suite *AnnounceTestSuite) TestAnnounceTwice() {
 	suite.True(ok)
 	suite.Equal(announcingAccount.ID, boost.AccountID)
 
+	// Insert the boost-of status into the
+	// DB cache to emulate processor handling
+	boost.ID, _ = id.NewULIDFromTime(boost.CreatedAt)
+	suite.state.Caches.GTS.Status().Store(boost, func() error {
+		return nil
+	})
+
 	// only the URI will be set on the boosted status because it still needs to be dereferenced
 	suite.NotEmpty(boost.BoostOf.URI)
 
diff --git a/internal/federation/federatingdb/create.go b/internal/federation/federatingdb/create.go
index bf3e7f75d..ca87131fe 100644
--- a/internal/federation/federatingdb/create.go
+++ b/internal/federation/federatingdb/create.go
@@ -103,11 +103,11 @@ func (f *federatingDB) activityBlock(ctx context.Context, asType vocab.Type, rec
 
 	block.ID = id.NewULID()
 
-	if err := f.db.PutBlock(ctx, block); err != nil {
+	if err := f.state.DB.PutBlock(ctx, block); err != nil {
 		return fmt.Errorf("activityBlock: database error inserting block: %s", err)
 	}
 
-	f.fedWorker.Queue(messages.FromFederator{
+	f.state.Workers.EnqueueFederator(ctx, messages.FromFederator{
 		APObjectType:     ap.ActivityBlock,
 		APActivityType:   ap.ActivityCreate,
 		GTSModel:         block,
@@ -202,7 +202,7 @@ func (f *federatingDB) createNote(ctx context.Context, note vocab.ActivityStream
 			return nil
 		}
 		// pass the note iri into the processor and have it do the dereferencing instead of doing it here
-		f.fedWorker.Queue(messages.FromFederator{
+		f.state.Workers.EnqueueFederator(ctx, messages.FromFederator{
 			APObjectType:     ap.ObjectNote,
 			APActivityType:   ap.ActivityCreate,
 			APIri:            id.GetIRI(),
@@ -226,7 +226,7 @@ func (f *federatingDB) createNote(ctx context.Context, note vocab.ActivityStream
 	}
 	status.ID = statusID
 
-	if err := f.db.PutStatus(ctx, status); err != nil {
+	if err := f.state.DB.PutStatus(ctx, status); err != nil {
 		if errors.Is(err, db.ErrAlreadyExists) {
 			// the status already exists in the database, which means we've already handled everything else,
 			// so we can just return nil here and be done with it.
@@ -236,7 +236,7 @@ func (f *federatingDB) createNote(ctx context.Context, note vocab.ActivityStream
 		return fmt.Errorf("createNote: database error inserting status: %s", err)
 	}
 
-	f.fedWorker.Queue(messages.FromFederator{
+	f.state.Workers.EnqueueFederator(ctx, messages.FromFederator{
 		APObjectType:     ap.ObjectNote,
 		APActivityType:   ap.ActivityCreate,
 		GTSModel:         status,
@@ -263,11 +263,11 @@ func (f *federatingDB) activityFollow(ctx context.Context, asType vocab.Type, re
 
 	followRequest.ID = id.NewULID()
 
-	if err := f.db.Put(ctx, followRequest); err != nil {
+	if err := f.state.DB.Put(ctx, followRequest); err != nil {
 		return fmt.Errorf("activityFollow: database error inserting follow request: %s", err)
 	}
 
-	f.fedWorker.Queue(messages.FromFederator{
+	f.state.Workers.EnqueueFederator(ctx, messages.FromFederator{
 		APObjectType:     ap.ActivityFollow,
 		APActivityType:   ap.ActivityCreate,
 		GTSModel:         followRequest,
@@ -294,11 +294,11 @@ func (f *federatingDB) activityLike(ctx context.Context, asType vocab.Type, rece
 
 	fave.ID = id.NewULID()
 
-	if err := f.db.Put(ctx, fave); err != nil {
+	if err := f.state.DB.Put(ctx, fave); err != nil {
 		return fmt.Errorf("activityLike: database error inserting fave: %s", err)
 	}
 
-	f.fedWorker.Queue(messages.FromFederator{
+	f.state.Workers.EnqueueFederator(ctx, messages.FromFederator{
 		APObjectType:     ap.ActivityLike,
 		APActivityType:   ap.ActivityCreate,
 		GTSModel:         fave,
@@ -325,11 +325,11 @@ func (f *federatingDB) activityFlag(ctx context.Context, asType vocab.Type, rece
 
 	report.ID = id.NewULID()
 
-	if err := f.db.PutReport(ctx, report); err != nil {
+	if err := f.state.DB.PutReport(ctx, report); err != nil {
 		return fmt.Errorf("activityFlag: database error inserting report: %w", err)
 	}
 
-	f.fedWorker.Queue(messages.FromFederator{
+	f.state.Workers.EnqueueFederator(ctx, messages.FromFederator{
 		APObjectType:     ap.ActivityFlag,
 		APActivityType:   ap.ActivityCreate,
 		GTSModel:         report,
diff --git a/internal/federation/federatingdb/db.go b/internal/federation/federatingdb/db.go
index 24455a553..af4aceeeb 100644
--- a/internal/federation/federatingdb/db.go
+++ b/internal/federation/federatingdb/db.go
@@ -24,9 +24,7 @@ import (
 	"codeberg.org/gruf/go-mutexes"
 	"github.com/superseriousbusiness/activity/pub"
 	"github.com/superseriousbusiness/activity/streams/vocab"
-	"github.com/superseriousbusiness/gotosocial/internal/concurrency"
-	"github.com/superseriousbusiness/gotosocial/internal/db"
-	"github.com/superseriousbusiness/gotosocial/internal/messages"
+	"github.com/superseriousbusiness/gotosocial/internal/state"
 	"github.com/superseriousbusiness/gotosocial/internal/typeutils"
 )
 
@@ -43,17 +41,15 @@ type DB interface {
 // It doesn't care what the underlying implementation of the DB interface is, as long as it works.
 type federatingDB struct {
 	locks         mutexes.MutexMap
-	db            db.DB
-	fedWorker     *concurrency.WorkerPool[messages.FromFederator]
+	state         *state.State
 	typeConverter typeutils.TypeConverter
 }
 
 // New returns a DB interface using the given database and config
-func New(db db.DB, fedWorker *concurrency.WorkerPool[messages.FromFederator], tc typeutils.TypeConverter) DB {
+func New(state *state.State, tc typeutils.TypeConverter) DB {
 	fdb := federatingDB{
 		locks:         mutexes.NewMap(-1, -1), // use defaults
-		db:            db,
-		fedWorker:     fedWorker,
+		state:         state,
 		typeConverter: tc,
 	}
 	return &fdb
diff --git a/internal/federation/federatingdb/delete.go b/internal/federation/federatingdb/delete.go
index a1890b56b..695f199b4 100644
--- a/internal/federation/federatingdb/delete.go
+++ b/internal/federation/federatingdb/delete.go
@@ -51,9 +51,9 @@ func (f *federatingDB) Delete(ctx context.Context, id *url.URL) error {
 
 	// 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.db.GetStatusByURI(ctx, id.String()); err == nil && requestingAccount.ID == s.AccountID {
+	if s, err := f.state.DB.GetStatusByURI(ctx, id.String()); err == nil && requestingAccount.ID == s.AccountID {
 		l.Debugf("uri is for STATUS with id: %s", s.ID)
-		f.fedWorker.Queue(messages.FromFederator{
+		f.state.Workers.EnqueueFederator(ctx, messages.FromFederator{
 			APObjectType:     ap.ObjectNote,
 			APActivityType:   ap.ActivityDelete,
 			GTSModel:         s,
@@ -61,9 +61,9 @@ func (f *federatingDB) Delete(ctx context.Context, id *url.URL) error {
 		})
 	}
 
-	if a, err := f.db.GetAccountByURI(ctx, id.String()); err == nil && requestingAccount.ID == a.ID {
+	if a, err := f.state.DB.GetAccountByURI(ctx, id.String()); err == nil && requestingAccount.ID == a.ID {
 		l.Debugf("uri is for ACCOUNT with id %s", a.ID)
-		f.fedWorker.Queue(messages.FromFederator{
+		f.state.Workers.EnqueueFederator(ctx, messages.FromFederator{
 			APObjectType:     ap.ObjectProfile,
 			APActivityType:   ap.ActivityDelete,
 			GTSModel:         a,
diff --git a/internal/federation/federatingdb/federatingdb_test.go b/internal/federation/federatingdb/federatingdb_test.go
index dd5a5f5f9..b0893f246 100644
--- a/internal/federation/federatingdb/federatingdb_test.go
+++ b/internal/federation/federatingdb/federatingdb_test.go
@@ -23,11 +23,11 @@ import (
 
 	"github.com/stretchr/testify/suite"
 	"github.com/superseriousbusiness/gotosocial/internal/ap"
-	"github.com/superseriousbusiness/gotosocial/internal/concurrency"
 	"github.com/superseriousbusiness/gotosocial/internal/db"
 	"github.com/superseriousbusiness/gotosocial/internal/federation/federatingdb"
 	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
 	"github.com/superseriousbusiness/gotosocial/internal/messages"
+	"github.com/superseriousbusiness/gotosocial/internal/state"
 	"github.com/superseriousbusiness/gotosocial/internal/typeutils"
 	"github.com/superseriousbusiness/gotosocial/testrig"
 )
@@ -36,9 +36,9 @@ type FederatingDBTestSuite struct {
 	suite.Suite
 	db            db.DB
 	tc            typeutils.TypeConverter
-	fedWorker     *concurrency.WorkerPool[messages.FromFederator]
 	fromFederator chan messages.FromFederator
 	federatingDB  federatingdb.DB
+	state         state.State
 
 	testTokens       map[string]*gtsmodel.Token
 	testClients      map[string]*gtsmodel.Client
@@ -66,22 +66,33 @@ func (suite *FederatingDBTestSuite) SetupTest() {
 	testrig.InitTestConfig()
 	testrig.InitTestLog()
 
-	suite.fedWorker = concurrency.NewWorkerPool[messages.FromFederator](-1, -1)
+	suite.state.Caches.Init()
+	testrig.StartWorkers(&suite.state)
+
 	suite.fromFederator = make(chan messages.FromFederator, 10)
-	suite.fedWorker.SetProcessor(func(ctx context.Context, msg messages.FromFederator) error {
+	suite.state.Workers.EnqueueFederator = func(ctx context.Context, msg messages.FromFederator) {
 		suite.fromFederator <- msg
-		return nil
-	})
-	_ = suite.fedWorker.Start()
-	suite.db = testrig.NewTestDB()
+	}
+
+	suite.db = testrig.NewTestDB(&suite.state)
 	suite.testActivities = testrig.NewTestActivities(suite.testAccounts)
 	suite.tc = testrig.NewTestTypeConverter(suite.db)
-	suite.federatingDB = testrig.NewTestFederatingDB(suite.db, suite.fedWorker)
+	suite.federatingDB = testrig.NewTestFederatingDB(&suite.state)
 	testrig.StandardDBSetup(suite.db, suite.testAccounts)
+
+	suite.state.DB = suite.db
 }
 
 func (suite *FederatingDBTestSuite) TearDownTest() {
 	testrig.StandardDBTeardown(suite.db)
+	testrig.StopWorkers(&suite.state)
+	for suite.fromFederator != nil {
+		select {
+		case <-suite.fromFederator:
+		default:
+			return
+		}
+	}
 }
 
 func createTestContext(receivingAccount *gtsmodel.Account, requestingAccount *gtsmodel.Account) context.Context {
diff --git a/internal/federation/federatingdb/followers.go b/internal/federation/federatingdb/followers.go
index c47a2b625..69746c99b 100644
--- a/internal/federation/federatingdb/followers.go
+++ b/internal/federation/federatingdb/followers.go
@@ -29,7 +29,7 @@ func (f *federatingDB) Followers(ctx context.Context, actorIRI *url.URL) (follow
 		return nil, err
 	}
 
-	acctFollowers, err := f.db.GetAccountFollowedBy(ctx, acct.ID, false)
+	acctFollowers, err := f.state.DB.GetAccountFollowedBy(ctx, acct.ID, false)
 	if err != nil {
 		return nil, fmt.Errorf("Followers: db error getting followers for account id %s: %s", acct.ID, err)
 	}
@@ -37,7 +37,7 @@ func (f *federatingDB) Followers(ctx context.Context, actorIRI *url.URL) (follow
 	iris := []*url.URL{}
 	for _, follow := range acctFollowers {
 		if follow.Account == nil {
-			a, err := f.db.GetAccountByID(ctx, follow.AccountID)
+			a, err := f.state.DB.GetAccountByID(ctx, follow.AccountID)
 			if err != nil {
 				errWrapped := fmt.Errorf("Followers: db error getting account id %s: %s", follow.AccountID, err)
 				if err == db.ErrNoEntries {
diff --git a/internal/federation/federatingdb/following.go b/internal/federation/federatingdb/following.go
index f4f07bb25..9c22c0574 100644
--- a/internal/federation/federatingdb/following.go
+++ b/internal/federation/federatingdb/following.go
@@ -47,7 +47,7 @@ func (f *federatingDB) Following(ctx context.Context, actorIRI *url.URL) (follow
 		return nil, err
 	}
 
-	acctFollowing, err := f.db.GetAccountFollows(ctx, acct.ID)
+	acctFollowing, err := f.state.DB.GetAccountFollows(ctx, acct.ID)
 	if err != nil {
 		return nil, fmt.Errorf("Following: db error getting following for account id %s: %s", acct.ID, err)
 	}
@@ -55,7 +55,7 @@ func (f *federatingDB) Following(ctx context.Context, actorIRI *url.URL) (follow
 	iris := []*url.URL{}
 	for _, follow := range acctFollowing {
 		if follow.TargetAccount == nil {
-			a, err := f.db.GetAccountByID(ctx, follow.TargetAccountID)
+			a, err := f.state.DB.GetAccountByID(ctx, follow.TargetAccountID)
 			if err != nil {
 				errWrapped := fmt.Errorf("Following: db error getting account id %s: %s", follow.TargetAccountID, err)
 				if err == db.ErrNoEntries {
diff --git a/internal/federation/federatingdb/get.go b/internal/federation/federatingdb/get.go
index 92a79d70f..1d687f110 100644
--- a/internal/federation/federatingdb/get.go
+++ b/internal/federation/federatingdb/get.go
@@ -39,13 +39,13 @@ func (f *federatingDB) Get(ctx context.Context, id *url.URL) (value vocab.Type,
 
 	switch {
 	case uris.IsUserPath(id):
-		acct, err := f.db.GetAccountByURI(ctx, id.String())
+		acct, err := f.state.DB.GetAccountByURI(ctx, id.String())
 		if err != nil {
 			return nil, err
 		}
 		return f.typeConverter.AccountToAS(ctx, acct)
 	case uris.IsStatusesPath(id):
-		status, err := f.db.GetStatusByURI(ctx, id.String())
+		status, err := f.state.DB.GetStatusByURI(ctx, id.String())
 		if err != nil {
 			return nil, err
 		}
diff --git a/internal/federation/federatingdb/inbox.go b/internal/federation/federatingdb/inbox.go
index 5ec735bd4..1a6da4ef0 100644
--- a/internal/federation/federatingdb/inbox.go
+++ b/internal/federation/federatingdb/inbox.go
@@ -85,12 +85,12 @@ func (f *federatingDB) InboxesForIRI(c context.Context, iri *url.URL) (inboxIRIs
 			return nil, fmt.Errorf("couldn't extract local account username from uri %s: %s", iri, err)
 		}
 
-		account, err := f.db.GetAccountByUsernameDomain(c, localAccountUsername, "")
+		account, err := f.state.DB.GetAccountByUsernameDomain(c, localAccountUsername, "")
 		if err != nil {
 			return nil, fmt.Errorf("couldn't find local account with username %s: %s", localAccountUsername, err)
 		}
 
-		follows, err := f.db.GetAccountFollowedBy(c, account.ID, false)
+		follows, err := f.state.DB.GetAccountFollowedBy(c, account.ID, false)
 		if err != nil {
 			return nil, fmt.Errorf("couldn't get followers of local account %s: %s", localAccountUsername, err)
 		}
@@ -98,7 +98,7 @@ func (f *federatingDB) InboxesForIRI(c context.Context, iri *url.URL) (inboxIRIs
 		for _, follow := range follows {
 			// make sure we retrieved the following account from the db
 			if follow.Account == nil {
-				followingAccount, err := f.db.GetAccountByID(c, follow.AccountID)
+				followingAccount, err := f.state.DB.GetAccountByID(c, follow.AccountID)
 				if err != nil {
 					if err == db.ErrNoEntries {
 						continue
@@ -126,7 +126,7 @@ func (f *federatingDB) InboxesForIRI(c context.Context, iri *url.URL) (inboxIRIs
 	}
 
 	// check if this is just an account IRI...
-	if account, err := f.db.GetAccountByURI(c, iri.String()); err == nil {
+	if account, err := f.state.DB.GetAccountByURI(c, iri.String()); err == nil {
 		// deliver to a shared inbox if we have that option
 		var inbox string
 		if config.GetInstanceDeliverToSharedInboxes() && account.SharedInboxURI != nil && *account.SharedInboxURI != "" {
diff --git a/internal/federation/federatingdb/owns.go b/internal/federation/federatingdb/owns.go
index def0fa518..2c11e8148 100644
--- a/internal/federation/federatingdb/owns.go
+++ b/internal/federation/federatingdb/owns.go
@@ -54,7 +54,7 @@ func (f *federatingDB) Owns(ctx context.Context, id *url.URL) (bool, error) {
 		if err != nil {
 			return false, fmt.Errorf("error parsing statuses path for url %s: %s", id.String(), err)
 		}
-		status, err := f.db.GetStatusByURI(ctx, uid)
+		status, err := f.state.DB.GetStatusByURI(ctx, uid)
 		if err != nil {
 			if err == db.ErrNoEntries {
 				// there are no entries for this status
@@ -71,7 +71,7 @@ func (f *federatingDB) Owns(ctx context.Context, id *url.URL) (bool, error) {
 		if err != nil {
 			return false, fmt.Errorf("error parsing statuses path for url %s: %s", id.String(), err)
 		}
-		if _, err := f.db.GetAccountByUsernameDomain(ctx, username, ""); err != nil {
+		if _, err := f.state.DB.GetAccountByUsernameDomain(ctx, username, ""); err != nil {
 			if err == db.ErrNoEntries {
 				// there are no entries for this username
 				return false, nil
@@ -88,7 +88,7 @@ func (f *federatingDB) Owns(ctx context.Context, id *url.URL) (bool, error) {
 		if err != nil {
 			return false, fmt.Errorf("error parsing statuses path for url %s: %s", id.String(), err)
 		}
-		if _, err := f.db.GetAccountByUsernameDomain(ctx, username, ""); err != nil {
+		if _, err := f.state.DB.GetAccountByUsernameDomain(ctx, username, ""); err != nil {
 			if err == db.ErrNoEntries {
 				// there are no entries for this username
 				return false, nil
@@ -105,7 +105,7 @@ func (f *federatingDB) Owns(ctx context.Context, id *url.URL) (bool, error) {
 		if err != nil {
 			return false, fmt.Errorf("error parsing statuses path for url %s: %s", id.String(), err)
 		}
-		if _, err := f.db.GetAccountByUsernameDomain(ctx, username, ""); err != nil {
+		if _, err := f.state.DB.GetAccountByUsernameDomain(ctx, username, ""); err != nil {
 			if err == db.ErrNoEntries {
 				// there are no entries for this username
 				return false, nil
@@ -122,7 +122,7 @@ func (f *federatingDB) Owns(ctx context.Context, id *url.URL) (bool, error) {
 		if err != nil {
 			return false, fmt.Errorf("error parsing like path for url %s: %s", id.String(), err)
 		}
-		if _, err := f.db.GetAccountByUsernameDomain(ctx, username, ""); err != nil {
+		if _, err := f.state.DB.GetAccountByUsernameDomain(ctx, username, ""); err != nil {
 			if err == db.ErrNoEntries {
 				// there are no entries for this username
 				return false, nil
@@ -130,7 +130,7 @@ func (f *federatingDB) Owns(ctx context.Context, id *url.URL) (bool, error) {
 			// an actual error happened
 			return false, fmt.Errorf("database error fetching account with username %s: %s", username, err)
 		}
-		if err := f.db.GetByID(ctx, likeID, &gtsmodel.StatusFave{}); err != nil {
+		if err := f.state.DB.GetByID(ctx, likeID, &gtsmodel.StatusFave{}); err != nil {
 			if err == db.ErrNoEntries {
 				// there are no entries
 				return false, nil
@@ -147,7 +147,7 @@ func (f *federatingDB) Owns(ctx context.Context, id *url.URL) (bool, error) {
 		if err != nil {
 			return false, fmt.Errorf("error parsing block path for url %s: %s", id.String(), err)
 		}
-		if _, err := f.db.GetAccountByUsernameDomain(ctx, username, ""); err != nil {
+		if _, err := f.state.DB.GetAccountByUsernameDomain(ctx, username, ""); err != nil {
 			if err == db.ErrNoEntries {
 				// there are no entries for this username
 				return false, nil
@@ -155,7 +155,7 @@ func (f *federatingDB) Owns(ctx context.Context, id *url.URL) (bool, error) {
 			// an actual error happened
 			return false, fmt.Errorf("database error fetching account with username %s: %s", username, err)
 		}
-		if err := f.db.GetByID(ctx, blockID, &gtsmodel.Block{}); err != nil {
+		if err := f.state.DB.GetByID(ctx, blockID, &gtsmodel.Block{}); err != nil {
 			if err == db.ErrNoEntries {
 				// there are no entries
 				return false, nil
diff --git a/internal/federation/federatingdb/reject.go b/internal/federation/federatingdb/reject.go
index 3c3cd7c75..d443cd6cb 100644
--- a/internal/federation/federatingdb/reject.go
+++ b/internal/federation/federatingdb/reject.go
@@ -64,7 +64,7 @@ func (f *federatingDB) Reject(ctx context.Context, reject vocab.ActivityStreamsR
 			if uris.IsFollowPath(rejectedObjectIRI) {
 				// REJECT FOLLOW
 				gtsFollowRequest := &gtsmodel.FollowRequest{}
-				if err := f.db.GetWhere(ctx, []db.Where{{Key: "uri", Value: rejectedObjectIRI.String()}}, gtsFollowRequest); err != nil {
+				if err := f.state.DB.GetWhere(ctx, []db.Where{{Key: "uri", Value: rejectedObjectIRI.String()}}, gtsFollowRequest); err != nil {
 					return fmt.Errorf("Reject: couldn't get follow request with id %s from the database: %s", rejectedObjectIRI.String(), err)
 				}
 
@@ -73,7 +73,7 @@ func (f *federatingDB) Reject(ctx context.Context, reject vocab.ActivityStreamsR
 					return errors.New("Reject: follow object account and inbox account were not the same")
 				}
 
-				if _, err := f.db.RejectFollowRequest(ctx, gtsFollowRequest.AccountID, gtsFollowRequest.TargetAccountID); err != nil {
+				if _, err := f.state.DB.RejectFollowRequest(ctx, gtsFollowRequest.AccountID, gtsFollowRequest.TargetAccountID); err != nil {
 					return err
 				}
 
@@ -102,7 +102,7 @@ func (f *federatingDB) Reject(ctx context.Context, reject vocab.ActivityStreamsR
 			if gtsFollow.AccountID != receivingAccount.ID {
 				return errors.New("Reject: follow object account and inbox account were not the same")
 			}
-			if _, err := f.db.RejectFollowRequest(ctx, gtsFollow.AccountID, gtsFollow.TargetAccountID); err != nil {
+			if _, err := f.state.DB.RejectFollowRequest(ctx, gtsFollow.AccountID, gtsFollow.TargetAccountID); err != nil {
 				return err
 			}
 
diff --git a/internal/federation/federatingdb/undo.go b/internal/federation/federatingdb/undo.go
index b239aabb4..e33b365fa 100644
--- a/internal/federation/federatingdb/undo.go
+++ b/internal/federation/federatingdb/undo.go
@@ -81,11 +81,11 @@ func (f *federatingDB) Undo(ctx context.Context, undo vocab.ActivityStreamsUndo)
 				return errors.New("UNDO: follow object account and inbox account were not the same")
 			}
 			// delete any existing FOLLOW
-			if err := f.db.DeleteWhere(ctx, []db.Where{{Key: "uri", Value: gtsFollow.URI}}, &gtsmodel.Follow{}); err != nil {
+			if err := f.state.DB.DeleteWhere(ctx, []db.Where{{Key: "uri", Value: gtsFollow.URI}}, &gtsmodel.Follow{}); err != nil {
 				return fmt.Errorf("UNDO: db error removing follow: %s", err)
 			}
 			// delete any existing FOLLOW REQUEST
-			if err := f.db.DeleteWhere(ctx, []db.Where{{Key: "uri", Value: gtsFollow.URI}}, &gtsmodel.FollowRequest{}); err != nil {
+			if err := f.state.DB.DeleteWhere(ctx, []db.Where{{Key: "uri", Value: gtsFollow.URI}}, &gtsmodel.FollowRequest{}); err != nil {
 				return fmt.Errorf("UNDO: db error removing follow request: %s", err)
 			}
 			l.Debug("follow undone")
@@ -114,7 +114,7 @@ func (f *federatingDB) Undo(ctx context.Context, undo vocab.ActivityStreamsUndo)
 				return errors.New("UNDO: block object account and inbox account were not the same")
 			}
 			// delete any existing BLOCK
-			if err := f.db.DeleteBlockByURI(ctx, gtsBlock.URI); err != nil {
+			if err := f.state.DB.DeleteBlockByURI(ctx, gtsBlock.URI); err != nil {
 				return fmt.Errorf("UNDO: db error removing block: %s", err)
 			}
 			l.Debug("block undone")
diff --git a/internal/federation/federatingdb/update.go b/internal/federation/federatingdb/update.go
index 570729a31..bed5de4db 100644
--- a/internal/federation/federatingdb/update.go
+++ b/internal/federation/federatingdb/update.go
@@ -138,7 +138,7 @@ func (f *federatingDB) Update(ctx context.Context, asType vocab.Type) error {
 
 		// pass to the processor for further updating of eg., avatar/header, emojis
 		// the actual db insert/update will take place a bit later
-		f.fedWorker.Queue(messages.FromFederator{
+		f.state.Workers.EnqueueFederator(ctx, messages.FromFederator{
 			APObjectType:     ap.ObjectProfile,
 			APActivityType:   ap.ActivityUpdate,
 			GTSModel:         updatedAcct,
diff --git a/internal/federation/federatingdb/util.go b/internal/federation/federatingdb/util.go
index 64f32d39c..f63eb6dc9 100644
--- a/internal/federation/federatingdb/util.go
+++ b/internal/federation/federatingdb/util.go
@@ -95,7 +95,7 @@ func (f *federatingDB) NewID(ctx context.Context, t vocab.Type) (idURL *url.URL,
 				// take the IRI of the first actor we can find (there should only be one)
 				if iter.IsIRI() {
 					// if there's an error here, just use the fallback behavior -- we don't need to return an error here
-					if actorAccount, err := f.db.GetAccountByURI(ctx, iter.GetIRI().String()); err == nil {
+					if actorAccount, err := f.state.DB.GetAccountByURI(ctx, iter.GetIRI().String()); err == nil {
 						newID, err := id.NewRandomULID()
 						if err != nil {
 							return nil, err
@@ -238,7 +238,7 @@ func (f *federatingDB) getAccountForIRI(ctx context.Context, iri *url.URL) (*gts
 
 	switch {
 	case uris.IsUserPath(iri):
-		if acct, err = f.db.GetAccountByURI(ctx, iri.String()); err != nil {
+		if acct, err = f.state.DB.GetAccountByURI(ctx, iri.String()); err != nil {
 			if err == db.ErrNoEntries {
 				return nil, fmt.Errorf("no actor found that corresponds to uri %s", iri.String())
 			}
@@ -246,7 +246,7 @@ func (f *federatingDB) getAccountForIRI(ctx context.Context, iri *url.URL) (*gts
 		}
 		return acct, nil
 	case uris.IsInboxPath(iri):
-		if err = f.db.GetWhere(ctx, []db.Where{{Key: "inbox_uri", Value: iri.String()}}, acct); err != nil {
+		if err = f.state.DB.GetWhere(ctx, []db.Where{{Key: "inbox_uri", Value: iri.String()}}, acct); err != nil {
 			if err == db.ErrNoEntries {
 				return nil, fmt.Errorf("no actor found that corresponds to inbox %s", iri.String())
 			}
@@ -254,7 +254,7 @@ func (f *federatingDB) getAccountForIRI(ctx context.Context, iri *url.URL) (*gts
 		}
 		return acct, nil
 	case uris.IsOutboxPath(iri):
-		if err = f.db.GetWhere(ctx, []db.Where{{Key: "outbox_uri", Value: iri.String()}}, acct); err != nil {
+		if err = f.state.DB.GetWhere(ctx, []db.Where{{Key: "outbox_uri", Value: iri.String()}}, acct); err != nil {
 			if err == db.ErrNoEntries {
 				return nil, fmt.Errorf("no actor found that corresponds to outbox %s", iri.String())
 			}
@@ -262,7 +262,7 @@ func (f *federatingDB) getAccountForIRI(ctx context.Context, iri *url.URL) (*gts
 		}
 		return acct, nil
 	case uris.IsFollowersPath(iri):
-		if err = f.db.GetWhere(ctx, []db.Where{{Key: "followers_uri", Value: iri.String()}}, acct); err != nil {
+		if err = f.state.DB.GetWhere(ctx, []db.Where{{Key: "followers_uri", Value: iri.String()}}, acct); err != nil {
 			if err == db.ErrNoEntries {
 				return nil, fmt.Errorf("no actor found that corresponds to followers_uri %s", iri.String())
 			}
@@ -270,7 +270,7 @@ func (f *federatingDB) getAccountForIRI(ctx context.Context, iri *url.URL) (*gts
 		}
 		return acct, nil
 	case uris.IsFollowingPath(iri):
-		if err = f.db.GetWhere(ctx, []db.Where{{Key: "following_uri", Value: iri.String()}}, acct); err != nil {
+		if err = f.state.DB.GetWhere(ctx, []db.Where{{Key: "following_uri", Value: iri.String()}}, acct); err != nil {
 			if err == db.ErrNoEntries {
 				return nil, fmt.Errorf("no actor found that corresponds to following_uri %s", iri.String())
 			}
diff --git a/internal/federation/federatingprotocol_test.go b/internal/federation/federatingprotocol_test.go
index faa168a71..e66cd78cb 100644
--- a/internal/federation/federatingprotocol_test.go
+++ b/internal/federation/federatingprotocol_test.go
@@ -28,10 +28,8 @@ import (
 	"github.com/go-fed/httpsig"
 	"github.com/stretchr/testify/suite"
 	"github.com/superseriousbusiness/gotosocial/internal/ap"
-	"github.com/superseriousbusiness/gotosocial/internal/concurrency"
 	"github.com/superseriousbusiness/gotosocial/internal/federation"
 	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
-	"github.com/superseriousbusiness/gotosocial/internal/messages"
 	"github.com/superseriousbusiness/gotosocial/testrig"
 )
 
@@ -43,12 +41,10 @@ func (suite *FederatingProtocolTestSuite) TestPostInboxRequestBodyHook1() {
 	// the activity we're gonna use
 	activity := suite.testActivities["dm_for_zork"]
 
-	fedWorker := concurrency.NewWorkerPool[messages.FromFederator](-1, -1)
-
 	httpClient := testrig.NewMockHTTPClient(nil, "../../testrig/media")
-	tc := testrig.NewTestTransportController(httpClient, suite.db, fedWorker)
+	tc := testrig.NewTestTransportController(&suite.state, httpClient)
 	// setup module being tested
-	federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(suite.db, fedWorker), tc, suite.tc, testrig.NewTestMediaManager(suite.db, suite.storage))
+	federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(&suite.state), tc, suite.tc, testrig.NewTestMediaManager(&suite.state))
 
 	// setup request
 	ctx := context.Background()
@@ -74,13 +70,11 @@ func (suite *FederatingProtocolTestSuite) TestPostInboxRequestBodyHook2() {
 	// the activity we're gonna use
 	activity := suite.testActivities["reply_to_turtle_for_zork"]
 
-	fedWorker := concurrency.NewWorkerPool[messages.FromFederator](-1, -1)
-
 	httpClient := testrig.NewMockHTTPClient(nil, "../../testrig/media")
-	tc := testrig.NewTestTransportController(httpClient, suite.db, fedWorker)
+	tc := testrig.NewTestTransportController(&suite.state, httpClient)
 
 	// setup module being tested
-	federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(suite.db, fedWorker), tc, suite.tc, testrig.NewTestMediaManager(suite.db, suite.storage))
+	federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(&suite.state), tc, suite.tc, testrig.NewTestMediaManager(&suite.state))
 
 	// setup request
 	ctx := context.Background()
@@ -107,13 +101,11 @@ func (suite *FederatingProtocolTestSuite) TestPostInboxRequestBodyHook3() {
 	// the activity we're gonna use
 	activity := suite.testActivities["reply_to_turtle_for_turtle"]
 
-	fedWorker := concurrency.NewWorkerPool[messages.FromFederator](-1, -1)
-
 	httpClient := testrig.NewMockHTTPClient(nil, "../../testrig/media")
-	tc := testrig.NewTestTransportController(httpClient, suite.db, fedWorker)
+	tc := testrig.NewTestTransportController(&suite.state, httpClient)
 
 	// setup module being tested
-	federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(suite.db, fedWorker), tc, suite.tc, testrig.NewTestMediaManager(suite.db, suite.storage))
+	federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(&suite.state), tc, suite.tc, testrig.NewTestMediaManager(&suite.state))
 
 	// setup request
 	ctx := context.Background()
@@ -142,13 +134,11 @@ func (suite *FederatingProtocolTestSuite) TestAuthenticatePostInbox() {
 	sendingAccount := suite.testAccounts["remote_account_1"]
 	inboxAccount := suite.testAccounts["local_account_1"]
 
-	fedWorker := concurrency.NewWorkerPool[messages.FromFederator](-1, -1)
-
 	httpClient := testrig.NewMockHTTPClient(nil, "../../testrig/media")
-	tc := testrig.NewTestTransportController(httpClient, suite.db, fedWorker)
+	tc := testrig.NewTestTransportController(&suite.state, httpClient)
 
 	// now setup module being tested, with the mock transport controller
-	federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(suite.db, fedWorker), tc, suite.tc, testrig.NewTestMediaManager(suite.db, suite.storage))
+	federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(&suite.state), tc, suite.tc, testrig.NewTestMediaManager(&suite.state))
 
 	request := httptest.NewRequest(http.MethodPost, "http://localhost:8080/users/the_mighty_zork/inbox", nil)
 	// we need these headers for the request to be validated
@@ -187,13 +177,11 @@ func (suite *FederatingProtocolTestSuite) TestAuthenticatePostGone() {
 	activity := suite.testActivities["delete_https://somewhere.mysterious/users/rest_in_piss#main-key"]
 	inboxAccount := suite.testAccounts["local_account_1"]
 
-	fedWorker := concurrency.NewWorkerPool[messages.FromFederator](-1, -1)
-
 	httpClient := testrig.NewMockHTTPClient(nil, "../../testrig/media")
-	tc := testrig.NewTestTransportController(httpClient, suite.db, fedWorker)
+	tc := testrig.NewTestTransportController(&suite.state, httpClient)
 
 	// now setup module being tested, with the mock transport controller
-	federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(suite.db, fedWorker), tc, suite.tc, testrig.NewTestMediaManager(suite.db, suite.storage))
+	federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(&suite.state), tc, suite.tc, testrig.NewTestMediaManager(&suite.state))
 
 	request := httptest.NewRequest(http.MethodPost, "http://localhost:8080/users/the_mighty_zork/inbox", nil)
 	// we need these headers for the request to be validated
@@ -231,13 +219,11 @@ func (suite *FederatingProtocolTestSuite) TestAuthenticatePostGoneNoTombstoneYet
 	activity := suite.testActivities["delete_https://somewhere.mysterious/users/rest_in_piss#main-key"]
 	inboxAccount := suite.testAccounts["local_account_1"]
 
-	fedWorker := concurrency.NewWorkerPool[messages.FromFederator](-1, -1)
-
 	httpClient := testrig.NewMockHTTPClient(nil, "../../testrig/media")
-	tc := testrig.NewTestTransportController(httpClient, suite.db, fedWorker)
+	tc := testrig.NewTestTransportController(&suite.state, httpClient)
 
 	// now setup module being tested, with the mock transport controller
-	federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(suite.db, fedWorker), tc, suite.tc, testrig.NewTestMediaManager(suite.db, suite.storage))
+	federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(&suite.state), tc, suite.tc, testrig.NewTestMediaManager(&suite.state))
 
 	request := httptest.NewRequest(http.MethodPost, "http://localhost:8080/users/the_mighty_zork/inbox", nil)
 	// we need these headers for the request to be validated
@@ -271,10 +257,9 @@ func (suite *FederatingProtocolTestSuite) TestAuthenticatePostGoneNoTombstoneYet
 }
 
 func (suite *FederatingProtocolTestSuite) TestBlocked1() {
-	fedWorker := concurrency.NewWorkerPool[messages.FromFederator](-1, -1)
 	httpClient := testrig.NewMockHTTPClient(nil, "../../testrig/media")
-	tc := testrig.NewTestTransportController(httpClient, suite.db, fedWorker)
-	federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(suite.db, fedWorker), tc, suite.tc, testrig.NewTestMediaManager(suite.db, suite.storage))
+	tc := testrig.NewTestTransportController(&suite.state, httpClient)
+	federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(&suite.state), tc, suite.tc, testrig.NewTestMediaManager(&suite.state))
 
 	sendingAccount := suite.testAccounts["remote_account_1"]
 	inboxAccount := suite.testAccounts["local_account_1"]
@@ -294,10 +279,9 @@ func (suite *FederatingProtocolTestSuite) TestBlocked1() {
 }
 
 func (suite *FederatingProtocolTestSuite) TestBlocked2() {
-	fedWorker := concurrency.NewWorkerPool[messages.FromFederator](-1, -1)
 	httpClient := testrig.NewMockHTTPClient(nil, "../../testrig/media")
-	tc := testrig.NewTestTransportController(httpClient, suite.db, fedWorker)
-	federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(suite.db, fedWorker), tc, suite.tc, testrig.NewTestMediaManager(suite.db, suite.storage))
+	tc := testrig.NewTestTransportController(&suite.state, httpClient)
+	federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(&suite.state), tc, suite.tc, testrig.NewTestMediaManager(&suite.state))
 
 	sendingAccount := suite.testAccounts["remote_account_1"]
 	inboxAccount := suite.testAccounts["local_account_1"]
@@ -328,10 +312,9 @@ func (suite *FederatingProtocolTestSuite) TestBlocked2() {
 }
 
 func (suite *FederatingProtocolTestSuite) TestBlocked3() {
-	fedWorker := concurrency.NewWorkerPool[messages.FromFederator](-1, -1)
 	httpClient := testrig.NewMockHTTPClient(nil, "../../testrig/media")
-	tc := testrig.NewTestTransportController(httpClient, suite.db, fedWorker)
-	federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(suite.db, fedWorker), tc, suite.tc, testrig.NewTestMediaManager(suite.db, suite.storage))
+	tc := testrig.NewTestTransportController(&suite.state, httpClient)
+	federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(&suite.state), tc, suite.tc, testrig.NewTestMediaManager(&suite.state))
 
 	sendingAccount := suite.testAccounts["remote_account_1"]
 	inboxAccount := suite.testAccounts["local_account_1"]
@@ -365,10 +348,9 @@ func (suite *FederatingProtocolTestSuite) TestBlocked3() {
 }
 
 func (suite *FederatingProtocolTestSuite) TestBlocked4() {
-	fedWorker := concurrency.NewWorkerPool[messages.FromFederator](-1, -1)
 	httpClient := testrig.NewMockHTTPClient(nil, "../../testrig/media")
-	tc := testrig.NewTestTransportController(httpClient, suite.db, fedWorker)
-	federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(suite.db, fedWorker), tc, suite.tc, testrig.NewTestMediaManager(suite.db, suite.storage))
+	tc := testrig.NewTestTransportController(&suite.state, httpClient)
+	federator := federation.NewFederator(suite.db, testrig.NewTestFederatingDB(&suite.state), tc, suite.tc, testrig.NewTestMediaManager(&suite.state))
 
 	sendingAccount := suite.testAccounts["remote_account_1"]
 	inboxAccount := suite.testAccounts["local_account_1"]
diff --git a/internal/federation/federator_test.go b/internal/federation/federator_test.go
index da6038ace..8a045aa1f 100644
--- a/internal/federation/federator_test.go
+++ b/internal/federation/federator_test.go
@@ -23,6 +23,7 @@ import (
 
 	"github.com/superseriousbusiness/gotosocial/internal/db"
 	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
+	"github.com/superseriousbusiness/gotosocial/internal/state"
 	"github.com/superseriousbusiness/gotosocial/internal/storage"
 	"github.com/superseriousbusiness/gotosocial/internal/typeutils"
 	"github.com/superseriousbusiness/gotosocial/testrig"
@@ -32,6 +33,7 @@ type FederatorStandardTestSuite struct {
 	suite.Suite
 	db             db.DB
 	storage        *storage.Driver
+	state          state.State
 	tc             typeutils.TypeConverter
 	testAccounts   map[string]*gtsmodel.Account
 	testStatuses   map[string]*gtsmodel.Status
@@ -42,8 +44,9 @@ type FederatorStandardTestSuite struct {
 // SetupSuite sets some variables on the suite that we can use as consts (more or less) throughout
 func (suite *FederatorStandardTestSuite) SetupSuite() {
 	// setup standard items
+	testrig.StartWorkers(&suite.state)
 	suite.storage = testrig.NewInMemoryStorage()
-	suite.tc = testrig.NewTestTypeConverter(suite.db)
+	suite.state.Storage = suite.storage
 	suite.testAccounts = testrig.NewTestAccounts()
 	suite.testStatuses = testrig.NewTestStatuses()
 	suite.testTombstones = testrig.NewTestTombstones()
@@ -52,7 +55,10 @@ func (suite *FederatorStandardTestSuite) SetupSuite() {
 func (suite *FederatorStandardTestSuite) SetupTest() {
 	testrig.InitTestConfig()
 	testrig.InitTestLog()
-	suite.db = testrig.NewTestDB()
+	suite.state.Caches.Init()
+	suite.db = testrig.NewTestDB(&suite.state)
+	suite.tc = testrig.NewTestTypeConverter(suite.db)
+	suite.state.DB = suite.db
 	suite.testActivities = testrig.NewTestActivities(suite.testAccounts)
 	testrig.StandardDBSetup(suite.db, suite.testAccounts)
 }