chore(server): split trigger worker into 3 workers (#3120)

* chore(server): split trigger worker into 3 workers

* rename runner file and delete unit test

* fix test

* fix(trigger-worker): trace-id injection

* address sebastian's comment

* fix: instrumentation

server/app/test_pipeline.go

@@ -68,16 +68,28 @@ func buildTestPipeline(
 		eventEmitter,
 	)
 
-	runner := executor.NewPersistentRunner(
+	triggerResolverWorker := executor.NewTriggerResolverWorker(
 		triggerRegistry,
 		execTestUpdater,
 		tracer,
-		subscriptionManager,
 		tracedbFactory,
 		dsRepo,
 		eventEmitter,
 	)
 
+	triggerExecuterWorker := executor.NewTriggerExecuterWorker(
+		triggerRegistry,
+		execTestUpdater,
+		tracer,
+		eventEmitter,
+	)
+
+	triggerResultProcessorWorker := executor.NewTriggerResultProcessorWorker(
+		tracer,
+		subscriptionManager,
+		eventEmitter,
+	)
+
 	cancelRunHandlerFn := executor.HandleRunCancelation(execTestUpdater, tracer, eventEmitter)
 
 	queueBuilder := executor.NewQueueBuilder().
@@ -92,7 +104,9 @@ func buildTestPipeline(
 	pgQueue := executor.NewPostgresQueueDriver(pool)
 
 	pipeline := executor.NewPipeline(queueBuilder,
-		executor.PipelineStep{Processor: runner, Driver: pgQueue.Channel("runner")},
+		executor.PipelineStep{Processor: triggerResolverWorker, Driver: pgQueue.Channel("trigger_resolve")},
+		executor.PipelineStep{Processor: triggerExecuterWorker, Driver: pgQueue.Channel("trigger_execute")},
+		executor.PipelineStep{Processor: triggerResultProcessorWorker, Driver: pgQueue.Channel("trigger_result")},
 		executor.PipelineStep{Processor: tracePoller, Driver: pgQueue.Channel("tracePoller")},
 		executor.PipelineStep{Processor: linterRunner, Driver: pgQueue.Channel("linterRunner")},
 		executor.PipelineStep{Processor: assertionRunner, Driver: pgQueue.Channel("assertionRunner")},

server/executor/poller_executor_test.go

@@ -26,7 +26,8 @@ import (
 )
 
 var (
-	randomIDGenerator = id.NewRandGenerator()
+	randomIDGenerator       = id.NewRandGenerator()
+	noError           error = nil
 )
 
 func Test_PollerExecutor_ExecuteRequest_NoRootSpan_NoSpanCase(t *testing.T) {