feat: Implement storage job limiter

internal/pkg/service/stream/sink/type/tablesink/keboola/bridge/bridge.go

@@ -105,10 +105,10 @@ func (b *Bridge) isKeboolaStagingFile(file *model.File) bool {
 	return file.StagingStorage.Provider == stagingFileProvider
 }
 
-func (b *Bridge) MirrorJobs(d dependencies) error {
+func (b *Bridge) MirrorJobs(ctx context.Context, d dependencies) error {
 	// Mirror jobs
 	wg := &sync.WaitGroup{}
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancel(ctx)
 	d.Process().OnShutdown(func(_ context.Context) {
 		b.logger.Info(ctx, "closing bridge job mirror")
 

internal/pkg/service/stream/storage/node/coordinator/coordinator.go

@@ -18,7 +18,7 @@ func Start(ctx context.Context, d dependencies.CoordinatorScope, cfg config.Conf
 	logger := d.Logger().WithComponent("storage.node.coordinator")
 	logger.Info(ctx, `starting storage coordinator node`)
 	// Change default bridge behaviour to use Mirror as it is required for sink throttling
-	err := d.KeboolaSinkBridge().MirrorJobs(d)
+	err := d.KeboolaSinkBridge().MirrorJobs(ctx, d)
 	if err != nil {
 		return err
 	}