[performance] processing media and scheduled jobs improvements (#1482)

* replace media workers with just runners.WorkerPool, move to state structure, use go-sched for global task scheduling

* improved code comment

* fix worker tryUntil function, update go-runners/go-sched

* make preprocess functions package public, use these where possible to stop doubled up processing

* remove separate emoji worker pool

* limit calls to time.Now() during media preprocessing

* use Processor{} to manage singular runtime of processing media

* ensure workers get started when media manager is used

* improved error setting in processing media, fix media test

* port changes from processingmedia to processing emoji

* finish code commenting

* finish code commenting and comment-out client API + federator worker pools until concurrency worker pools replaced

* linterrrrrrrrrrrrrrrr

---------

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

1 files changed, 18 insertions, 16 deletions

diff --git a/cmd/gotosocial/action/server/server.go b/cmd/gotosocial/action/server/server.go
index 74a5be24b..89486a5dc 100644
--- a/cmd/gotosocial/action/server/server.go
+++ b/cmd/gotosocial/action/server/server.go
@@ -91,33 +91,35 @@ var Start action.GTSAction = func(ctx context.Context) error {
 		return fmt.Errorf("error creating instance instance: %s", err)
 	}
 
-	// Create the client API and federator worker pools
-	// NOTE: these MUST NOT be used until they are passed to the
-	// processor and it is started. The reason being that the processor
-	// sets the Worker process functions and start the underlying pools
-	clientWorker := concurrency.NewWorkerPool[messages.FromClientAPI](-1, -1)
-	fedWorker := concurrency.NewWorkerPool[messages.FromFederator](-1, -1)
-
-	federatingDB := federatingdb.New(dbService, fedWorker)
-
-	// build converters and util
-	typeConverter := typeutils.NewConverter(dbService)
-
 	// Open the storage backend
 	storage, err := gtsstorage.AutoConfig()
 	if err != nil {
 		return fmt.Errorf("error creating storage backend: %w", err)
 	}
 
+	// Set the state storage driver
+	state.Storage = storage
+
 	// Build HTTP client (TODO: add configurables here)
 	client := httpclient.New(httpclient.Config{})
 
+	// Initialize workers.
+	state.Workers.Start()
+	defer state.Workers.Stop()
+
+	// Create the client API and federator worker pools
+	// NOTE: these MUST NOT be used until they are passed to the
+	// processor and it is started. The reason being that the processor
+	// sets the Worker process functions and start the underlying pools
+	// TODO: move these into state.Workers (and maybe reformat worker pools).
+	clientWorker := concurrency.NewWorkerPool[messages.FromClientAPI](-1, -1)
+	fedWorker := concurrency.NewWorkerPool[messages.FromFederator](-1, -1)
+
 	// build backend handlers
-	mediaManager, err := media.NewManager(dbService, storage)
-	if err != nil {
-		return fmt.Errorf("error creating media manager: %s", err)
-	}
+	mediaManager := media.NewManager(&state)
 	oauthServer := oauth.New(ctx, dbService)
+	typeConverter := typeutils.NewConverter(dbService)
+	federatingDB := federatingdb.New(dbService, fedWorker, typeConverter)
 	transportController := transport.NewController(dbService, federatingDB, &federation.Clock{}, client)
 	federator := federation.NewFederator(dbService, federatingDB, transportController, typeConverter, mediaManager)