feat(sdk): restart goroutines (#5821)

Signed-off-by: francois  samin <[email protected]>

cli/cdsctl/events.go

@@ -60,8 +60,8 @@ func eventsListenRun(v cli.Values) error {
 	chanMessageToSend := make(chan []sdk.WebsocketFilter)
 	chanErrorReceived := make(chan error)
 
-	sdk.NewGoRoutines().Run(ctx, "WebsocketEventsListenCmd", func(ctx context.Context) {
-		client.WebsocketEventsListen(ctx, sdk.NewGoRoutines(), chanMessageToSend, chanMessageReceived, chanErrorReceived)
+	sdk.NewGoRoutines(ctx).Run(ctx, "WebsocketEventsListenCmd", func(ctx context.Context) {
+		client.WebsocketEventsListen(ctx, sdk.NewGoRoutines(ctx), chanMessageToSend, chanMessageReceived, chanErrorReceived)
 	})
 
 	switch {

cli/cdsctl/workflow_log.go

@@ -437,7 +437,7 @@ func workflowLogStreamRun(v cli.Values) error {
 	chanMsgReceived := make(chan json.RawMessage)
 	chanErrorReceived := make(chan error)
 
-	goRoutines := sdk.NewGoRoutines()
+	goRoutines := sdk.NewGoRoutines(ctx)
 	goRoutines.Exec(ctx, "WebsocketEventsListenCmd", func(ctx context.Context) {
 		for ctx.Err() == nil {
 			if err := client.RequestWebsocket(ctx, goRoutines, fmt.Sprintf("%s/item/stream", link.CDNURL), chanMessageToSend, chanMsgReceived, chanErrorReceived); err != nil {

cli/cdsctl/workflow_transform_as_code.go

@@ -53,8 +53,8 @@ func workflowTransformAsCodeRun(v cli.Values) (interface{}, error) {
 	chanMessageToSend := make(chan []sdk.WebsocketFilter)
 	chanErrorReceived := make(chan error)
 
-	sdk.NewGoRoutines().Run(ctx, "WebsocketEventsListenCmd", func(ctx context.Context) {
-		client.WebsocketEventsListen(ctx, sdk.NewGoRoutines(), chanMessageToSend, chanMessageReceived, chanErrorReceived)
+	sdk.NewGoRoutines(ctx).Run(ctx, "WebsocketEventsListenCmd", func(ctx context.Context) {
+		client.WebsocketEventsListen(ctx, sdk.NewGoRoutines(ctx), chanMessageToSend, chanMessageReceived, chanErrorReceived)
 	})
 
 	ope, err := client.WorkflowTransformAsCode(projectKey, v.GetString(_WorkflowName), branch, message)

engine/api/api.go

@@ -545,7 +545,7 @@ func (a *API) Serve(ctx context.Context) error {
 	}
 
 	log.Info(ctx, "Initializing HTTP router")
-	a.GoRoutines = sdk.NewGoRoutines()
+	a.GoRoutines = sdk.NewGoRoutines(ctx)
 	a.Router = &Router{
 		Mux:        mux.NewRouter(),
 		Background: ctx,
@@ -652,12 +652,12 @@ func (a *API) Serve(ctx context.Context) error {
 		log.Error(ctx, "error while initializing event system: %s", err)
 	}
 
-	a.GoRoutines.Run(ctx, "event.dequeue", func(ctx context.Context) {
+	a.GoRoutines.RunWithRestart(ctx, "event.dequeue", func(ctx context.Context) {
 		event.DequeueEvent(ctx, a.mustDB())
 	})
 
 	log.Info(ctx, "Initializing internal routines...")
-	a.GoRoutines.Run(ctx, "maintenance.Subscribe", func(ctx context.Context) {
+	a.GoRoutines.RunWithRestart(ctx, "maintenance.Subscribe", func(ctx context.Context) {
 		if err := a.listenMaintenance(ctx); err != nil {
 			log.Error(ctx, "error while initializing listen maintenance routine: %s", err)
 		}
@@ -668,7 +668,7 @@ func (a *API) Serve(ctx context.Context) error {
 			log.Error(ctx, "error while initializing worker models routine: %s", err)
 		}
 	})
-	a.GoRoutines.Run(ctx, "worker.Initialize", func(ctx context.Context) {
+	a.GoRoutines.RunWithRestart(ctx, "worker.Initialize", func(ctx context.Context) {
 		if err := worker.Initialize(ctx, a.DBConnectionFactory.GetDBMap(gorpmapping.Mapper), a.Cache); err != nil {
 			log.Error(ctx, "error while initializing workers routine: %s", err)
 		}
@@ -684,25 +684,25 @@ func (a *API) Serve(ctx context.Context) error {
 	a.GoRoutines.Run(ctx, "audit.ComputeWorkflowAudit", func(ctx context.Context) {
 		audit.ComputeWorkflowAudit(ctx, a.DBConnectionFactory.GetDBMap(gorpmapping.Mapper))
 	})
-	a.GoRoutines.Run(ctx, "auditCleanerRoutine(ctx", func(ctx context.Context) {
+	a.GoRoutines.Run(ctx, "auditCleanerRoutine", func(ctx context.Context) {
 		auditCleanerRoutine(ctx, a.DBConnectionFactory.GetDBMap(gorpmapping.Mapper))
 	})
-	a.GoRoutines.Run(ctx, "repositoriesmanager.ReceiveEvents", func(ctx context.Context) {
+	a.GoRoutines.RunWithRestart(ctx, "repositoriesmanager.ReceiveEvents", func(ctx context.Context) {
 		repositoriesmanager.ReceiveEvents(ctx, a.DBConnectionFactory.GetDBMap(gorpmapping.Mapper), a.Cache)
 	})
-	a.GoRoutines.Run(ctx, "services.KillDeadServices", func(ctx context.Context) {
+	a.GoRoutines.RunWithRestart(ctx, "services.KillDeadServices", func(ctx context.Context) {
 		services.KillDeadServices(ctx, a.mustDB)
 	})
 	a.GoRoutines.Run(ctx, "broadcast.Initialize", func(ctx context.Context) {
 		broadcast.Initialize(ctx, a.DBConnectionFactory.GetDBMap(gorpmapping.Mapper))
 	})
-	a.GoRoutines.Run(ctx, "api.serviceAPIHeartbeat", func(ctx context.Context) {
+	a.GoRoutines.RunWithRestart(ctx, "api.serviceAPIHeartbeat", func(ctx context.Context) {
 		a.serviceAPIHeartbeat(ctx)
 	})
-	a.GoRoutines.Run(ctx, "authentication.SessionCleaner", func(ctx context.Context) {
+	a.GoRoutines.RunWithRestart(ctx, "authentication.SessionCleaner", func(ctx context.Context) {
 		authentication.SessionCleaner(ctx, a.mustDB, 10*time.Second)
 	})
-	a.GoRoutines.Run(ctx, "api.WorkflowRunCraft", func(ctx context.Context) {
+	a.GoRoutines.RunWithRestart(ctx, "api.WorkflowRunCraft", func(ctx context.Context) {
 		a.WorkflowRunCraft(ctx, 100*time.Millisecond)
 	})
 

engine/api/api_test.go

@@ -42,7 +42,7 @@ func newTestAPI(t *testing.T, bootstrapFunc ...test.Bootstrapf) (*API, *test.Fak
 	api.AuthenticationDrivers[sdk.ConsumerBuiltin] = builtin.NewDriver()
 	api.AuthenticationDrivers[sdk.ConsumerTest] = authdrivertest.NewDriver(t)
 	api.AuthenticationDrivers[sdk.ConsumerTest2] = authdrivertest.NewDriver(t)
-	api.GoRoutines = sdk.NewGoRoutines()
+	api.GoRoutines = sdk.NewGoRoutines(context.TODO())
 
 	api.InitRouter()
 
@@ -90,7 +90,7 @@ func newTestServer(t *testing.T, bootstrapFunc ...test.Bootstrapf) (*API, *test.
 	api.AuthenticationDrivers = make(map[sdk.AuthConsumerType]sdk.AuthDriver)
 	api.AuthenticationDrivers[sdk.ConsumerLocal] = local.NewDriver(context.TODO(), false, "http://localhost:8080", "")
 	api.AuthenticationDrivers[sdk.ConsumerBuiltin] = builtin.NewDriver()
-	api.GoRoutines = sdk.NewGoRoutines()
+	api.GoRoutines = sdk.NewGoRoutines(context.TODO())
 
 	api.InitRouter()
 	ts := httptest.NewServer(router.Mux)

engine/api/application_test.go

@@ -274,7 +274,7 @@ func TestUpdateAsCodeApplicationHandler(t *testing.T) {
 	chanMessageReceived := make(chan sdk.WebsocketEvent)
 	chanMessageToSend := make(chan []sdk.WebsocketFilter)
 	chanErrorReceived := make(chan error)
-	go client.WebsocketEventsListen(context.TODO(), sdk.NewGoRoutines(), chanMessageToSend, chanMessageReceived, chanErrorReceived)
+	go client.WebsocketEventsListen(context.TODO(), sdk.NewGoRoutines(context.TODO()), chanMessageToSend, chanMessageReceived, chanErrorReceived)
 	chanMessageToSend <- []sdk.WebsocketFilter{{
 		Type:         sdk.WebsocketFilterTypeAscodeEvent,
 		ProjectKey:   proj.Key,

engine/api/environment_ascode_test.go

@@ -233,7 +233,7 @@ func TestUpdateAsCodeEnvironmentHandler(t *testing.T) {
 	chanMessageReceived := make(chan sdk.WebsocketEvent)
 	chanMessageToSend := make(chan []sdk.WebsocketFilter)
 	chanErrorReceived := make(chan error)
-	go client.WebsocketEventsListen(context.TODO(), sdk.NewGoRoutines(), chanMessageToSend, chanMessageReceived, chanErrorReceived)
+	go client.WebsocketEventsListen(context.TODO(), sdk.NewGoRoutines(context.TODO()), chanMessageToSend, chanMessageReceived, chanErrorReceived)
 	chanMessageToSend <- []sdk.WebsocketFilter{{
 		Type:         sdk.WebsocketFilterTypeAscodeEvent,
 		ProjectKey:   proj.Key,

engine/api/migrate/migration.go

@@ -46,7 +46,7 @@ func Run(ctx context.Context, db gorp.SqlExecutor) {
 				wg.Add(1)
 			}
 
-			sdk.NewGoRoutines().Run(ctx, "migrate_"+currentMigration.Name, func(contex context.Context) {
+			sdk.NewGoRoutines(ctx).Run(ctx, "migrate_"+currentMigration.Name, func(contex context.Context) {
 				defer func() {
 					if currentMigration.Blocker {
 						wg.Done()

engine/api/pipeline_test.go

@@ -205,7 +205,7 @@ func TestUpdateAsCodePipelineHandler(t *testing.T) {
 	chanMessageToSend := make(chan []sdk.WebsocketFilter)
 	chanErrorReceived := make(chan error)
 
-	go client.WebsocketEventsListen(context.TODO(), sdk.NewGoRoutines(), chanMessageToSend, chanMessageReceived, chanErrorReceived)
+	go client.WebsocketEventsListen(context.TODO(), sdk.NewGoRoutines(context.TODO()), chanMessageToSend, chanMessageReceived, chanErrorReceived)
 	chanMessageToSend <- []sdk.WebsocketFilter{{
 		Type:         sdk.WebsocketFilterTypeAscodeEvent,
 		ProjectKey:   proj.Key,

engine/api/websocket_test.go

@@ -45,7 +45,7 @@ func Test_websocketWrongFilters(t *testing.T) {
 		InsecureSkipVerifyTLS:             true,
 		BuitinConsumerAuthenticationToken: jws,
 	})
-	go client.WebsocketEventsListen(context.TODO(), sdk.NewGoRoutines(), chanMessageToSend, chanMessageReceived, chanErrorReceived)
+	go client.WebsocketEventsListen(context.TODO(), sdk.NewGoRoutines(context.TODO()), chanMessageToSend, chanMessageReceived, chanErrorReceived)
 
 	// Subscribe to project without project key
 	chanMessageToSend <- []sdk.WebsocketFilter{{
@@ -126,7 +126,7 @@ func Test_websocketGetWorkflowEvent(t *testing.T) {
 		InsecureSkipVerifyTLS: true,
 		SessionToken:          jwt,
 	})
-	go client.WebsocketEventsListen(context.TODO(), sdk.NewGoRoutines(), chanMessageToSend, chanMessageReceived, chanErrorReceived)
+	go client.WebsocketEventsListen(context.TODO(), sdk.NewGoRoutines(context.TODO()), chanMessageToSend, chanMessageReceived, chanErrorReceived)
 	var lastResponse *sdk.WebsocketEvent
 	go func() {
 		for e := range chanMessageReceived {
@@ -268,7 +268,7 @@ func TestWebsocketNoEventLoose(t *testing.T) {
 		InsecureSkipVerifyTLS: true,
 		SessionToken:          jwt,
 	})
-	go client1.WebsocketEventsListen(context.TODO(), sdk.NewGoRoutines(), chan1MessageToSend, chan1MessageReceived, chan1ErrorReceived)
+	go client1.WebsocketEventsListen(context.TODO(), sdk.NewGoRoutines(ctx), chan1MessageToSend, chan1MessageReceived, chan1ErrorReceived)
 	var client1EventCount int64
 	go func() {
 		for {
@@ -299,7 +299,7 @@ func TestWebsocketNoEventLoose(t *testing.T) {
 		SessionToken:          jwt,
 	})
 	var client2EventCount int64
-	go client2.WebsocketEventsListen(context.TODO(), sdk.NewGoRoutines(), chan2MessageToSend, chan2MessageReceived, chan2ErrorReceived)
+	go client2.WebsocketEventsListen(context.TODO(), sdk.NewGoRoutines(ctx), chan2MessageToSend, chan2MessageReceived, chan2ErrorReceived)
 	go func() {
 		for {
 			select {

engine/api/workflow_ascode_test.go

@@ -211,7 +211,7 @@ func TestPostUpdateWorkflowAsCodeHandler(t *testing.T) {
 	chanMessageReceived := make(chan sdk.WebsocketEvent)
 	chanMessageToSend := make(chan []sdk.WebsocketFilter)
 	chanErrorReceived := make(chan error)
-	go client.WebsocketEventsListen(context.TODO(), sdk.NewGoRoutines(), chanMessageToSend, chanMessageReceived, chanErrorReceived)
+	go client.WebsocketEventsListen(context.TODO(), sdk.NewGoRoutines(context.TODO()), chanMessageToSend, chanMessageReceived, chanErrorReceived)
 	chanMessageToSend <- []sdk.WebsocketFilter{{
 		Type:         sdk.WebsocketFilterTypeAscodeEvent,
 		ProjectKey:   proj.Key,
@@ -422,7 +422,7 @@ func TestPostMigrateWorkflowAsCodeHandler(t *testing.T) {
 	chanMessageReceived := make(chan sdk.WebsocketEvent)
 	chanMessageToSend := make(chan []sdk.WebsocketFilter)
 	chanErrorReceived := make(chan error)
-	go client.WebsocketEventsListen(context.TODO(), sdk.NewGoRoutines(), chanMessageToSend, chanMessageReceived, chanErrorReceived)
+	go client.WebsocketEventsListen(context.TODO(), sdk.NewGoRoutines(context.TODO()), chanMessageToSend, chanMessageReceived, chanErrorReceived)
 	chanMessageToSend <- []sdk.WebsocketFilter{{
 		Type:         sdk.WebsocketFilterTypeAscodeEvent,
 		ProjectKey:   proj.Key,

engine/api/workflow_purge_test.go

@@ -98,7 +98,7 @@ func Test_purgeDryRunHandler(t *testing.T) {
 	})
 	contextWS, cancel := context.WithCancel(context.Background())
 	t.Cleanup(cancel)
-	go client.WebsocketEventsListen(contextWS, sdk.NewGoRoutines(), chanMessageToSend, chanMessageReceived, chanErrorReceived)
+	go client.WebsocketEventsListen(contextWS, sdk.NewGoRoutines(context.TODO()), chanMessageToSend, chanMessageReceived, chanErrorReceived)
 
 	// Subscribe to workflow retention
 	chanMessageToSend <- []sdk.WebsocketFilter{{

engine/api/workflow_run_test.go

@@ -1407,7 +1407,7 @@ func Test_postWorkflowRunAsyncFailedHandler(t *testing.T) {
 		OperationUUID: ope.UUID,
 	}
 
-	ascode.UpdateAsCodeResult(context.TODO(), api.mustDB(), api.Cache, sdk.NewGoRoutines(), *proj, *w1, app, ed, u)
+	ascode.UpdateAsCodeResult(context.TODO(), api.mustDB(), api.Cache, sdk.NewGoRoutines(context.TODO()), *proj, *w1, app, ed, u)
 
 	// Prepare request
 	uri := router.GetRoute("POST", api.postWorkflowRunHandler, map[string]string{

engine/cdn/cdn.go

@@ -36,8 +36,7 @@ const (
 // New returns a new service
 func New() *Service {
 	s := new(Service)
-	s.GoRoutines = sdk.NewGoRoutines()
-
+	s.GoRoutines = sdk.NewGoRoutines(context.Background())
 	return s
 }
 

engine/cdn/cdn_gc_test.go

@@ -39,15 +39,15 @@ func TestCleanSynchronizedItem(t *testing.T) {
 		Cache:               cache,
 		Mapper:              m,
 	}
-	s.GoRoutines = sdk.NewGoRoutines()
+	s.GoRoutines = sdk.NewGoRoutines(context.TODO())
 
 	tmpDir, err := ioutil.TempDir("", t.Name()+"-cdn-1-*")
 	require.NoError(t, err)
 
 	ctx, cancel := context.WithTimeout(context.TODO(), 5*time.Second)
 	t.Cleanup(cancel)
 
-	cdnUnits, err := storage.Init(ctx, m, cache, db.DbMap, sdk.NewGoRoutines(), storage.Configuration{
+	cdnUnits, err := storage.Init(ctx, m, cache, db.DbMap, sdk.NewGoRoutines(ctx), storage.Configuration{
 		HashLocatorSalt: "thisismysalt",
 		Buffers: map[string]storage.BufferConfiguration{
 			"redis_buffer": {
@@ -208,15 +208,15 @@ func TestCleanSynchronizedItemWithDisabledStorage(t *testing.T) {
 		Cache:               cache,
 		Mapper:              m,
 	}
-	s.GoRoutines = sdk.NewGoRoutines()
+	s.GoRoutines = sdk.NewGoRoutines(context.TODO())
 
 	tmpDir, err := ioutil.TempDir("", t.Name()+"-cdn-1-*")
 	require.NoError(t, err)
 
 	ctx, cancel := context.WithTimeout(context.TODO(), 5*time.Second)
 	t.Cleanup(cancel)
 
-	cdnUnits, err := storage.Init(ctx, m, cache, db.DbMap, sdk.NewGoRoutines(), storage.Configuration{
+	cdnUnits, err := storage.Init(ctx, m, cache, db.DbMap, sdk.NewGoRoutines(ctx), storage.Configuration{
 		HashLocatorSalt: "thisismysalt",
 		Buffers: map[string]storage.BufferConfiguration{
 			"redis_buffer": {
@@ -351,7 +351,7 @@ func TestCleanWaitingItem(t *testing.T) {
 		Cache:               cache,
 		Mapper:              m,
 	}
-	s.GoRoutines = sdk.NewGoRoutines()
+	s.GoRoutines = sdk.NewGoRoutines(context.TODO())
 
 	ctx, cancel := context.WithCancel(context.TODO())
 	t.Cleanup(cancel)
@@ -402,7 +402,7 @@ func TestCleanWaitingItemWithoutItemUnit(t *testing.T) {
 		Cache:               cache,
 		Mapper:              m,
 	}
-	s.GoRoutines = sdk.NewGoRoutines()
+	s.GoRoutines = sdk.NewGoRoutines(context.TODO())
 
 	ctx, cancel := context.WithCancel(context.TODO())
 	t.Cleanup(cancel)
@@ -446,7 +446,7 @@ func TestPurgeItem(t *testing.T) {
 		Cache:               cache,
 		Mapper:              m,
 	}
-	s.GoRoutines = sdk.NewGoRoutines()
+	s.GoRoutines = sdk.NewGoRoutines(context.TODO())
 
 	ctx, cancel := context.WithCancel(context.TODO())
 	t.Cleanup(cancel)

engine/cdn/cdn_item_test.go

@@ -41,7 +41,7 @@ func TestGetItemValue(t *testing.T) {
 		Cache:               cache,
 		Mapper:              m,
 	}
-	s.GoRoutines = sdk.NewGoRoutines()
+	s.GoRoutines = sdk.NewGoRoutines(context.TODO())
 
 	ctx, cancel := context.WithCancel(context.TODO())
 	t.Cleanup(cancel)
@@ -202,7 +202,7 @@ func TestGetItemValue_ThousandLines(t *testing.T) {
 		Cache:               cache,
 		Mapper:              m,
 	}
-	s.GoRoutines = sdk.NewGoRoutines()
+	s.GoRoutines = sdk.NewGoRoutines(context.TODO())
 
 	ctx, cancel := context.WithCancel(context.TODO())
 	t.Cleanup(cancel)
@@ -309,7 +309,7 @@ func TestGetItemValue_Reverse(t *testing.T) {
 		Cache:               cache,
 		Mapper:              m,
 	}
-	s.GoRoutines = sdk.NewGoRoutines()
+	s.GoRoutines = sdk.NewGoRoutines(context.TODO())
 
 	ctx, cancel := context.WithCancel(context.TODO())
 	t.Cleanup(cancel)
@@ -419,7 +419,7 @@ func TestGetItemValue_ThousandLinesReverse(t *testing.T) {
 		Mapper:              m,
 	}
 	s.Cfg.Log.StepMaxSize = 200000
-	s.GoRoutines = sdk.NewGoRoutines()
+	s.GoRoutines = sdk.NewGoRoutines(context.TODO())
 
 	ctx, cancel := context.WithCancel(context.TODO())
 	t.Cleanup(cancel)

engine/cdn/cdn_log_store_test.go

@@ -40,7 +40,7 @@ func TestStoreNewStepLog(t *testing.T) {
 		Cache:               cache,
 		Mapper:              m,
 	}
-	s.GoRoutines = sdk.NewGoRoutines()
+	s.GoRoutines = sdk.NewGoRoutines(context.TODO())
 
 	ctx, cancel := context.WithCancel(context.TODO())
 	t.Cleanup(cancel)
@@ -123,7 +123,7 @@ func TestStoreLastStepLog(t *testing.T) {
 		Cache:               cache,
 		Mapper:              m,
 	}
-	s.GoRoutines = sdk.NewGoRoutines()
+	s.GoRoutines = sdk.NewGoRoutines(context.TODO())
 
 	ctx, cancel := context.WithCancel(context.TODO())
 	t.Cleanup(cancel)
@@ -212,7 +212,7 @@ func TestStoreNewServiceLog(t *testing.T) {
 		Cache:               cache,
 		Mapper:              m,
 	}
-	s.GoRoutines = sdk.NewGoRoutines()
+	s.GoRoutines = sdk.NewGoRoutines(context.TODO())
 
 	ctx, cancel := context.WithCancel(context.TODO())
 	t.Cleanup(cancel)

engine/cdn/cdn_log_tcp_test.go

@@ -61,7 +61,7 @@ func TestWorkerLogCDNEnabled(t *testing.T) {
 	}
 	tmpDir, err := ioutil.TempDir("", t.Name()+"-cdn-1-*")
 	require.NoError(t, err)
-	cdnUnits, err := storage.Init(context.TODO(), m, store, db.DbMap, sdk.NewGoRoutines(), storage.Configuration{
+	cdnUnits, err := storage.Init(context.TODO(), m, store, db.DbMap, sdk.NewGoRoutines(context.TODO()), storage.Configuration{
 		HashLocatorSalt: "thisismysalt",
 		Buffers: map[string]storage.BufferConfiguration{
 			"redis_buffer": {
@@ -84,7 +84,7 @@ func TestWorkerLogCDNEnabled(t *testing.T) {
 	s.Units = cdnUnits
 
 	s.Cfg.Log.StepMaxSize = 1000
-	s.GoRoutines = sdk.NewGoRoutines()
+	s.GoRoutines = sdk.NewGoRoutines(context.TODO())
 
 	signature := cdn.Signature{
 		Worker: &cdn.SignatureWorker{
@@ -177,10 +177,10 @@ func TestServiceLogCDNDisabled(t *testing.T) {
 		Mapper:              m,
 	}
 	s.Cfg.Log.StepMaxSize = 1000
-	s.GoRoutines = sdk.NewGoRoutines()
+	s.GoRoutines = sdk.NewGoRoutines(context.TODO())
 	tmpDir, err := ioutil.TempDir("", t.Name()+"-cdn-1-*")
 	require.NoError(t, err)
-	cdnUnits, err := storage.Init(context.TODO(), m, store, db.DbMap, sdk.NewGoRoutines(), storage.Configuration{
+	cdnUnits, err := storage.Init(context.TODO(), m, store, db.DbMap, sdk.NewGoRoutines(context.TODO()), storage.Configuration{
 		HashLocatorSalt: "thisismysalt",
 		Buffers: map[string]storage.BufferConfiguration{
 			"redis_buffer": {
@@ -292,7 +292,7 @@ func TestStoreTruncatedLogs(t *testing.T) {
 		Cache:               cache,
 		Mapper:              m,
 	}
-	s.GoRoutines = sdk.NewGoRoutines()
+	s.GoRoutines = sdk.NewGoRoutines(context.TODO())
 
 	ctx, ccl := context.WithCancel(context.TODO())
 	t.Cleanup(ccl)